src/cuga/backend/tools_env/registry/mcp_manager/README.md

MCPManager

A Python utility to load multiple OpenAPI schemas, spin up local SSE MCP servers, register their tools, and invoke them asynchronously.

Features

• Fetch and parse OpenAPI definitions (JSON or YAML)
• Spin up a per-schema SSE server on a free port
• Dynamically register each tool/function with metadata
• Call any registered tool by name with parameters and headers
• Inspect available APIs and response schemas

Example

1. Configure
   Define a list of schema names and URLs:

   schema_list = [
     {"petstore": "https://petstore3.swagger.io/api/v3/openapi.json"},
     {"myapi":   "https://example.com/path/to/openapi.yaml"}
   ]
   

2. Initialize & Run

   import asyncio
   from mcp_manager import MCPManager
   
   async def main():
       manager = MCPManager(schema_list)
       await manager.initialize_servers()
       await manager.run_all_servers()
       # List available servers
       print(manager.get_server_names())
       # Call a tool
       result = await manager.call_tool('petstore_findpetsbystatus', {"status": "available"})
       print(result)
   
   if __name__ == "__main__":
       asyncio.run(main())
   

Methods

MCPManager(schema_urls: List[Dict[str, str]])

Constructor. schema_urls is a list of single-key dicts mapping server name to OpenAPI URL.

initialize_servers() -> None

Fetches each OpenAPI spec, builds a parser, creates an MCP server, and registers its tools.

run_all_servers() -> None

Starts each MCP server in a background thread using SSE transport.

call_tool(tool_name: str, args: dict, headers: dict = None) -> Any

Invoke a registered tool by name. Returns the JSON-decoded response.

get_server_names() -> List[str]

List all initialized server names.

get_all_apis(include_response_schema: bool=False) -> Dict[str, List[dict]]

Return all registered tools for each server. If include_response_schema is True, each tool dict gets a responseSchema field.