File size: 1,907 Bytes
915668d |
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 |
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()
# Ensure the response is in the expected format
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() |