---
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)
```