| from strands.tools.mcp import MCPClient |
| from mcp.client.sse import sse_client |
| from config.settings import settings |
| import logging |
| from typing import List, Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class MCPService: |
| """Centralized MCP service for managing Topcoder MCP client""" |
| |
| _instance = None |
| |
| def __new__(cls): |
| if cls._instance is None: |
| cls._instance = super(MCPService, cls).__new__(cls) |
| return cls._instance |
| |
| def create_topcoder_client(self) -> MCPClient: |
| """Create a Topcoder MCP client instance""" |
| logger.info(f"Creating Topcoder MCP client for endpoint: {settings.TOPCODER_MCP_ENDPOINT}") |
| return MCPClient(lambda: sse_client(settings.TOPCODER_MCP_ENDPOINT)) |
| |
| def get_topcoder_tools(self, client: MCPClient) -> List[Any]: |
| """Get available Topcoder MCP tools from the client""" |
| try: |
| tools = client.list_tools_sync() |
| tool_names = [getattr(tool, 'name', str(tool)) for tool in tools] |
| logger.info(f"Available Topcoder MCP tools: {tool_names}") |
| return tools |
| except Exception as e: |
| logger.error(f"Error getting Topcoder MCP tools: {e}") |
| return [] |
|
|
| |
| mcp_service = MCPService() |
|
|