Initialize MCP tools (both local and remote). Should be called once during application startup, before creating any agents. Args: local_servers_dir: Directory for local MCP servers, defaults to tools/mcp/ remote_servers: List of remote server configurations, ea
(
local_servers_dir: Optional[str] = None,
remote_servers: Optional[List[Dict[str, Any]]] = None
)
| 34 | |
| 35 | |
| 36 | async def init_mcp_tools( |
| 37 | local_servers_dir: Optional[str] = None, |
| 38 | remote_servers: Optional[List[Dict[str, Any]]] = None |
| 39 | ) -> MCPManager: |
| 40 | """ |
| 41 | Initialize MCP tools (both local and remote). |
| 42 | |
| 43 | Should be called once during application startup, before creating any agents. |
| 44 | |
| 45 | Args: |
| 46 | local_servers_dir: Directory for local MCP servers, defaults to tools/mcp/ |
| 47 | remote_servers: List of remote server configurations, each containing: |
| 48 | - id: Unique server identifier |
| 49 | - url: Server URL endpoint |
| 50 | - headers: Optional dict of HTTP headers (for authentication) |
| 51 | |
| 52 | Returns: |
| 53 | MCPManager instance |
| 54 | """ |
| 55 | global _mcp_manager, _mcp_initialized |
| 56 | |
| 57 | if _mcp_initialized: |
| 58 | logger.warning("MCP tools already initialized") |
| 59 | return _mcp_manager |
| 60 | |
| 61 | if _mcp_manager is None: |
| 62 | _mcp_manager = MCPManager() |
| 63 | |
| 64 | # Initialize MCP connections (both local and remote) |
| 65 | await _mcp_manager.initialize(local_servers_dir, remote_servers) |
| 66 | |
| 67 | # Attach MCP manager to tool registry |
| 68 | registry = get_registry() |
| 69 | registry.set_mcp_manager(_mcp_manager) |
| 70 | |
| 71 | _mcp_initialized = True |
| 72 | logger.info("MCP tools initialized and attached to registry") |
| 73 | |
| 74 | return _mcp_manager |
| 75 | |
| 76 | |
| 77 | def get_mcp_manager() -> Optional[MCPManager]: |
no test coverage detected