| import httpx |
| from typing import List, Dict, Any |
|
|
| class MCPClient: |
| """ |
| A client for interacting with a Model Context Protocol (MCP) server. |
| Handles listing and executing tools via HTTP requests. |
| """ |
| def __init__(self, server_url: str): |
| self.server_url = server_url.rstrip('/') |
| self.http_client = httpx.AsyncClient(timeout=30.0) |
|
|
| async def list_tools(self) -> List[Dict[str, Any]]: |
| """Fetches the list of available tools from the MCP server.""" |
| try: |
| response = await self.http_client.get(f"{self.server_url}/mcp/tools") |
| response.raise_for_status() |
| tools_response = response.json() |
| |
| if "tools" in tools_response and isinstance(tools_response["tools"], list): |
| return tools_response["tools"] |
| return [] |
| except (httpx.RequestError, httpx.HTTPStatusError) as e: |
| print(f"Error fetching tools from {self.server_url}: {e}") |
| return [] |
|
|
| async def execute_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]: |
| """Executes a specific tool on the MCP server with the given parameters.""" |
| try: |
| response = await self.http_client.post( |
| f"{self.server_url}/mcp/tools/{tool_name}", |
| json={"params": params} |
| ) |
| response.raise_for_status() |
| return response.json() |
| except (httpx.RequestError, httpx.HTTPStatusError) as e: |
| print(f"Error executing tool '{tool_name}' on {self.server_url}: {e}") |
| return { |
| "status": "error", |
| "result": f"Failed to connect or execute tool on {self.server_url}. Error: {e}" |
| } |
|
|
| async def close(self): |
| """Closes the underlying HTTP client.""" |
| await self.http_client.aclose() |
