File size: 4,586 Bytes
67f8819 43149e3 67f8819 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """
MCP service wrapper for MCP server lifecycle management.
Per @specs/001-chatbot-mcp/plan.md Section VIII - AI Chatbot Architecture
MCP First: All task operations go through MCP SDK for OpenAI Agents integration.
This service manages the MCP server lifecycle including startup, shutdown,
and tool registration.
"""
from typing import Optional, Any, Dict
from contextlib import asynccontextmanager
import logging
from mcp.server import get_mcp_server
logger = logging.getLogger(__name__)
class MCPService:
"""
Service wrapper for MCP server lifecycle management.
This class provides a singleton interface to the MCP server,
managing its lifecycle and providing access to tool execution.
"""
_instance: Optional["MCPService"] = None
_server: Optional[Any] = None
def __new__(cls) -> "MCPService":
"""Singleton pattern to ensure only one MCP server instance."""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
async def initialize(self) -> None:
"""
Initialize the MCP server and register all tools.
This should be called during FastAPI application startup.
Per @specs/001-chatbot-mcp/plan.md, MCP server starts with the FastAPI app.
"""
if self._server is None:
try:
self._server = get_mcp_server()
logger.info("MCP server initialized successfully with todo-mcp-server")
except Exception as e:
logger.error(f"Failed to initialize MCP server: {e}")
raise
async def shutdown(self) -> None:
"""
Shutdown the MCP server gracefully.
This should be called during FastAPI application shutdown.
"""
if self._server is not None:
try:
# FastMCP handles cleanup automatically
logger.info("MCP server shut down successfully")
except Exception as e:
logger.error(f"Error during MCP server shutdown: {e}")
def get_server(self) -> Any:
"""
Get the MCP server instance.
Returns:
The FastMCP server instance
Raises:
RuntimeError: If server has not been initialized
"""
if self._server is None:
raise RuntimeError("MCP server not initialized. Call initialize() first.")
return self._server
async def list_tools(self) -> list[Dict[str, Any]]:
"""
List all available MCP tools.
Returns:
List of tool definitions with name, description, and schemas
Per @specs/001-chatbot-mcp/contracts/mcp-tools.json
"""
server = self.get_server()
# FastMCP stores tools in _tool_manager
if hasattr(server, '_tool_manager'):
tools = server._tool_manager.list_tools()
return [
{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema,
"output_schema": tool.outputSchema if hasattr(tool, 'outputSchema') else None
}
for tool in tools
]
return []
def get_tool_info(self) -> Dict[str, Any]:
"""
Get information about available MCP tools.
Returns:
Dict with tool count, server info, and tool names
Useful for debugging and monitoring.
"""
server = self.get_server()
return {
"server_name": getattr(server, 'name', 'todo-mcp-server'),
"initialized": self._server is not None,
"tool_count": len(server.list_tools()) if hasattr(server, 'list_tools') else 0
}
def _get_tool_names(self) -> list[str]:
"""Get list of registered tool names."""
server = self.get_server()
if hasattr(server, 'list_tools'):
tools = server.list_tools()
return [tool.name for tool in tools]
return []
# Singleton instance
mcp_service = MCPService()
@asynccontextmanager
async def mcp_lifespan():
"""
Async context manager for MCP server lifecycle.
Usage in FastAPI:
@asynccontextmanager
async def lifespan(app: FastAPI):
async with mcp_lifespan():
yield
Per @specs/001-chatbot-mcp/plan.md, MCP server lifecycle tied to FastAPI app.
"""
try:
await mcp_service.initialize()
yield
finally:
await mcp_service.shutdown()
|