> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# MCP Client Protocol
> Protocol interface for MCP client implementations
The `MCPClientProtocol` interface enables custom MCP client implementations while maintaining compatibility with the PraisonAI agent system.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "MCP Client Extensibility"
Protocol[📋 MCPClientProtocol] --> Custom[🔧 Custom Client]
Protocol --> Builtin[⚡ Built-in Client]
Custom --> Agent[🤖 Agent]
Builtin --> Agent
end
classDef protocol fill:#6366F1,stroke:#7C90A0,color:#fff
classDef implementation fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef result fill:#10B981,stroke:#7C90A0,color:#fff
class Protocol protocol
class Custom,Builtin implementation
class Agent result
```
## Quick Start
Create a custom MCP client that follows the protocol:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.mcp import MCPClientProtocol
from typing import Any, Dict, List
class MyCustomMCPClient:
def list_tools(self) -> List[Dict[str, Any]]:
return [
{
"name": "my_tool",
"description": "Custom tool implementation",
"inputSchema": {
"type": "object",
"properties": {
"message": {"type": "string"}
},
"required": ["message"]
}
}
]
def get_tools(self) -> List[Dict[str, Any]]:
return self.list_tools()
def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
if name == "my_tool":
return f"Custom response: {args.get('message', '')}"
raise ValueError(f"Unknown tool: {name}")
def shutdown(self) -> None:
print("Custom client shutting down")
# Verify protocol compliance
assert isinstance(MyCustomMCPClient(), MCPClientProtocol)
```
Wire your custom client into an agent:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
custom_client = MyCustomMCPClient()
agent = Agent(
name="custom-client-test",
instructions="Use the custom MCP client",
tools=custom_client
)
agent.start("Use my_tool to say hello")
```
***
## Protocol Methods
| Method | Sync | Async | Description |
| ----------------------- | ---- | ---------------------- | ------------------------------------------------- |
| `list_tools()` | ✅ | ✅ (`async_list_tools`) | List available tools from the server |
| `get_tools()` | ✅ | ✅ (`async_get_tools`) | Alias for `list_tools()` (backward compatibility) |
| `call_tool(name, args)` | ✅ | ✅ (`async_call_tool`) | Execute a tool with given arguments |
| `shutdown()` | ✅ | ✅ (`async_shutdown`) | Clean up resources and connections |
***
## Implementation Example
### HTTP-based MCP Client
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
from typing import Any, Dict, List
from praisonaiagents.mcp import MCPClientProtocol
class HTTPMCPClient:
def __init__(self, base_url: str, headers: Dict[str, str] = None):
self.base_url = base_url.rstrip('/')
self.headers = headers or {}
self.client = httpx.Client(headers=self.headers)
def list_tools(self) -> List[Dict[str, Any]]:
response = self.client.get(f"{self.base_url}/tools")
response.raise_for_status()
return response.json()["tools"]
def get_tools(self) -> List[Dict[str, Any]]:
return self.list_tools()
def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
response = self.client.post(
f"{self.base_url}/tools/{name}",
json={"arguments": args}
)
response.raise_for_status()
return response.json()["result"]
def shutdown(self) -> None:
self.client.close()
# Async versions
async def async_list_tools(self) -> List[Dict[str, Any]]:
async with httpx.AsyncClient(headers=self.headers) as client:
response = await client.get(f"{self.base_url}/tools")
response.raise_for_status()
return response.json()["tools"]
async def async_get_tools(self) -> List[Dict[str, Any]]:
return await self.async_list_tools()
async def async_call_tool(self, name: str, args: Dict[str, Any]) -> Any:
async with httpx.AsyncClient(headers=self.headers) as client:
response = await client.post(
f"{self.base_url}/tools/{name}",
json={"arguments": args}
)
response.raise_for_status()
return response.json()["result"]
async def async_shutdown(self) -> None:
# Cleanup async resources if needed
pass
# Usage
client = HTTPMCPClient("https://api.example.com/mcp")
agent = Agent(name="http-mcp", tools=client)
```
***
## Protocol Compliance
The protocol interface provides runtime type checking:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.mcp import MCPClientProtocol
# Check if your implementation follows the protocol
def validate_client(client):
if isinstance(client, MCPClientProtocol):
print("✅ Client follows MCPClientProtocol")
return True
else:
print("❌ Client does not implement required methods")
return False
# Test your implementation
my_client = MyCustomMCPClient()
validate_client(my_client) # ✅ Client follows MCPClientProtocol
```
### Required Methods
Every MCP client implementation must provide:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
class MinimalMCPClient:
def list_tools(self) -> List[Dict[str, Any]]: ...
def get_tools(self) -> List[Dict[str, Any]]: ...
def call_tool(self, name: str, args: Dict[str, Any]) -> Any: ...
def shutdown(self) -> None: ...
# Async versions (if supporting async)
async def async_list_tools(self) -> List[Dict[str, Any]]: ...
async def async_get_tools(self) -> List[Dict[str, Any]]: ...
async def async_call_tool(self, name: str, args: Dict[str, Any]) -> Any: ...
async def async_shutdown(self) -> None: ...
```
***
## Common Use Cases
### Database MCP Client
Connect to databases directly without external servers:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import sqlite3
from typing import Any, Dict, List
class DatabaseMCPClient:
def __init__(self, database_path: str):
self.db_path = database_path
def list_tools(self) -> List[Dict[str, Any]]:
return [
{
"name": "query_database",
"description": "Execute SQL query on database",
"inputSchema": {
"type": "object",
"properties": {
"sql": {"type": "string"}
},
"required": ["sql"]
}
}
]
def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
if name == "query_database":
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(args["sql"])
return cursor.fetchall()
```
### Mock MCP Client (Testing)
Create predictable responses for testing:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
class MockMCPClient:
def __init__(self, mock_tools: Dict[str, Any]):
self.mock_tools = mock_tools
def list_tools(self) -> List[Dict[str, Any]]:
return [
{"name": name, "description": f"Mock tool {name}"}
for name in self.mock_tools.keys()
]
def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
return self.mock_tools.get(name, f"Mock result for {name}")
# Usage in tests
mock_client = MockMCPClient({
"test_tool": "Expected test result",
"calculate": 42
})
```
***
## Best Practices
While not strictly required, providing both sync and async versions of methods ensures compatibility with different agent execution modes.
Implement proper error handling in `call_tool()`. Raise descriptive exceptions that help with debugging and user feedback.
Always implement `shutdown()` to clean up connections, files, or other resources. This prevents resource leaks in long-running applications.
Return proper JSON schemas in `list_tools()` to enable agent tool validation and better error messages.
***
## Related
Wire configured MCP servers into agents with one line
Restrict which MCP tools an agent can see and call