340 lines
No EOL
7.6 KiB
Text
340 lines
No EOL
7.6 KiB
Text
---
|
|
title: "Python Class Reference"
|
|
description: "Complete Python SDK class documentation"
|
|
---
|
|
|
|
## Server Class
|
|
|
|
The main class for implementing MCP servers in Python.
|
|
|
|
```python
|
|
from mcp_python.server import Server
|
|
```
|
|
|
|
### Constructor
|
|
|
|
```python
|
|
Server(name: str, version: str = "1.0.0")
|
|
```
|
|
|
|
Creates a new MCP server instance.
|
|
|
|
**Parameters:**
|
|
- `name` (str): Name of the server
|
|
- `version` (str, optional): Version of the server. Defaults to "1.0.0"
|
|
|
|
### Decorators
|
|
|
|
#### Resource Handling
|
|
|
|
##### @server.list_resources()
|
|
```python
|
|
@server.list_resources()
|
|
async def list_resources() -> List[Resource]:
|
|
"""Lists available resources."""
|
|
```
|
|
|
|
**Returns:** List of `Resource` objects containing:
|
|
- `uri` (str): Resource identifier
|
|
- `name` (str): Human-readable name
|
|
- `description` (str, optional): Resource description
|
|
- `mimeType` (str, optional): MIME type of the resource
|
|
|
|
##### @server.read_resource()
|
|
```python
|
|
@server.read_resource()
|
|
async def read_resource(uri: str) -> Union[str, bytes]:
|
|
"""Reads content of a specific resource."""
|
|
```
|
|
|
|
**Parameters:**
|
|
- `uri` (str): Resource URI to read
|
|
|
|
**Returns:** Content of the resource as either string or bytes
|
|
|
|
##### @server.subscribe_resource()
|
|
```python
|
|
@server.subscribe_resource()
|
|
async def subscribe_resource(uri: str) -> None:
|
|
"""Handles resource subscription requests."""
|
|
```
|
|
|
|
**Parameters:**
|
|
- `uri` (str): Resource URI to subscribe to
|
|
|
|
##### @server.unsubscribe_resource()
|
|
```python
|
|
@server.unsubscribe_resource()
|
|
async def unsubscribe_resource(uri: str) -> None:
|
|
"""Handles resource unsubscribe requests."""
|
|
```
|
|
|
|
**Parameters:**
|
|
- `uri` (str): Resource URI to unsubscribe from
|
|
|
|
#### Tool Support
|
|
|
|
##### @server.list_tools()
|
|
```python
|
|
@server.list_tools()
|
|
async def list_tools() -> List[Tool]:
|
|
"""Lists available tools."""
|
|
```
|
|
|
|
**Returns:** List of `Tool` objects containing:
|
|
- `name` (str): Tool identifier
|
|
- `description` (str): Human-readable description
|
|
- `parameters` (dict): JSON Schema defining tool parameters
|
|
|
|
##### @server.call_tool()
|
|
```python
|
|
@server.call_tool()
|
|
async def call_tool(name: str, **kwargs) -> Any:
|
|
"""Handles tool execution requests."""
|
|
```
|
|
|
|
**Parameters:**
|
|
- `name` (str): Tool identifier
|
|
- `**kwargs`: Tool-specific parameters
|
|
|
|
**Returns:** Tool execution result (any JSON-serializable value)
|
|
|
|
#### Prompt Management
|
|
|
|
##### @server.list_prompts()
|
|
```python
|
|
@server.list_prompts()
|
|
async def list_prompts() -> List[Prompt]:
|
|
"""Lists available prompts."""
|
|
```
|
|
|
|
**Returns:** List of `Prompt` objects containing:
|
|
- `name` (str): Prompt identifier
|
|
- `description` (str, optional): Human-readable description
|
|
- `parameters` (dict, optional): JSON Schema defining prompt parameters
|
|
|
|
##### @server.get_prompt()
|
|
```python
|
|
@server.get_prompt()
|
|
async def get_prompt(name: str, arguments: dict) -> PromptResponse:
|
|
"""Retrieves specific prompt templates."""
|
|
```
|
|
|
|
**Parameters:**
|
|
- `name` (str): Prompt identifier
|
|
- `arguments` (dict): Prompt parameters
|
|
|
|
**Returns:** `PromptResponse` object containing:
|
|
- `messages` (List[Message]): List of prompt messages
|
|
- `description` (str, optional): Description of the prompt response
|
|
|
|
### Session Methods
|
|
|
|
Methods available on the server session object:
|
|
|
|
#### send_resource_updated()
|
|
```python
|
|
async def send_resource_updated(uri: str) -> None:
|
|
"""Notifies clients that a resource has been updated."""
|
|
```
|
|
|
|
#### send_progress_notification()
|
|
```python
|
|
async def send_progress_notification(
|
|
token: str,
|
|
progress: int,
|
|
total: Optional[int] = None
|
|
) -> None:
|
|
"""Sends progress updates for long-running operations."""
|
|
```
|
|
|
|
#### send_log_message()
|
|
```python
|
|
async def send_log_message(
|
|
level: str,
|
|
data: Any,
|
|
logger: Optional[str] = None
|
|
) -> None:
|
|
"""Sends log messages to clients."""
|
|
```
|
|
|
|
## Data Classes
|
|
|
|
### Resource
|
|
```python
|
|
@dataclass
|
|
class Resource:
|
|
uri: str
|
|
name: str
|
|
description: Optional[str] = None
|
|
mimeType: Optional[str] = None
|
|
```
|
|
|
|
### Tool
|
|
```python
|
|
@dataclass
|
|
class Tool:
|
|
name: str
|
|
description: str
|
|
parameters: dict
|
|
```
|
|
|
|
### Prompt
|
|
```python
|
|
@dataclass
|
|
class Prompt:
|
|
name: str
|
|
description: Optional[str] = None
|
|
parameters: Optional[dict] = None
|
|
```
|
|
|
|
### Message
|
|
```python
|
|
@dataclass
|
|
class Message:
|
|
role: Literal["user", "assistant"]
|
|
content: Union[TextContent, ImageContent]
|
|
```
|
|
|
|
### PromptResponse
|
|
```python
|
|
@dataclass
|
|
class PromptResponse:
|
|
messages: List[Message]
|
|
description: Optional[str] = None
|
|
```
|
|
|
|
## Example Usage
|
|
|
|
Here's a complete example demonstrating the main features:
|
|
|
|
```python
|
|
from mcp_python.server import Server
|
|
import anyio
|
|
|
|
server = Server("example-server")
|
|
|
|
@server.list_resources()
|
|
async def list_resources():
|
|
return [
|
|
{
|
|
"uri": "example://resource1",
|
|
"name": "Example Resource 1",
|
|
"description": "An example resource"
|
|
}
|
|
]
|
|
|
|
@server.read_resource()
|
|
async def read_resource(uri):
|
|
if uri == "example://resource1":
|
|
return "Example resource content"
|
|
raise ValueError(f"Unknown resource: {uri}")
|
|
|
|
@server.list_tools()
|
|
async def list_tools():
|
|
return [
|
|
{
|
|
"name": "example-tool",
|
|
"description": "An example tool",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"param1": {"type": "string"}
|
|
}
|
|
}
|
|
}
|
|
]
|
|
|
|
@server.call_tool()
|
|
async def call_tool(name, **kwargs):
|
|
if name == "example-tool":
|
|
return f"Tool executed with params: {kwargs}"
|
|
raise ValueError(f"Unknown tool: {name}")
|
|
|
|
async def main():
|
|
async with stdio_server() as (read_stream, write_stream):
|
|
await server.run(
|
|
read_stream,
|
|
write_stream,
|
|
server.create_initialization_options()
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
anyio.run(main)
|
|
```
|
|
|
|
## Best Practices
|
|
|
|
1. **Error Handling**: Always implement proper error handling in your server methods:
|
|
```python
|
|
@server.read_resource()
|
|
async def read_resource(uri):
|
|
try:
|
|
return await fetch_resource(uri)
|
|
except ResourceNotFoundError as e:
|
|
raise ValueError(f"Resource not found: {uri}") from e
|
|
```
|
|
|
|
2. **Type Hints**: Use typing hints for better code clarity:
|
|
```python
|
|
@server.list_resources()
|
|
async def list_resources() -> List[Resource]:
|
|
resources: List[Resource] = []
|
|
return resources
|
|
```
|
|
|
|
3. **Progress Updates**: For long-running operations, send progress updates:
|
|
```python
|
|
@server.call_tool()
|
|
async def call_tool(name: str, session: ServerSession, **kwargs):
|
|
for i in range(100):
|
|
await session.send_progress_notification(
|
|
"operation-1",
|
|
progress=i,
|
|
total=100
|
|
)
|
|
```
|
|
|
|
4. **Resource Updates**: Notify clients when resources change:
|
|
```python
|
|
@server.call_tool()
|
|
async def call_tool(name: str, session: ServerSession, **kwargs):
|
|
# ... perform update ...
|
|
await session.send_resource_updated("example://resource1")
|
|
```
|
|
|
|
## Transport Options
|
|
|
|
### stdio Transport
|
|
```python
|
|
async with stdio_server() as (read_stream, write_stream):
|
|
await server.run(read_stream, write_stream)
|
|
```
|
|
|
|
### SSE Transport
|
|
```python
|
|
from mcp_python.transport import SSETransport
|
|
|
|
app = FastAPI()
|
|
transport = SSETransport()
|
|
|
|
@app.get("/sse")
|
|
async def sse_endpoint():
|
|
return await transport.handle_sse()
|
|
|
|
@app.post("/message")
|
|
async def message_endpoint(request: Request):
|
|
return await transport.handle_message(request)
|
|
```
|
|
|
|
### WebSocket Transport
|
|
```python
|
|
from mcp_python.transport import WebSocketTransport
|
|
|
|
app = FastAPI()
|
|
transport = WebSocketTransport()
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_endpoint(websocket: WebSocket):
|
|
await transport.handle_websocket(websocket)
|
|
``` |