> ## Documentation Index
> Fetch the complete documentation index at: https://docs.upsonic.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# MCPHandler

> Use MCPHandler class to connect to a single MCP server

## Overview

`MCPHandler` is the recommended way to connect to a single MCP server. It provides more control and flexibility than the class-based approach.

## Basic Usage

### Command String

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

# Create MCP handler with command string
mcp_handler = MCPHandler(
    command="uvx mcp-server-sqlite --db-path /tmp/db.db",
    timeout_seconds=60
)

# Create agent
agent = Agent(
    name="MCPHandler Command Agent",
    role="Database operations specialist using MCPHandler with command",
    goal="Demonstrate MCPHandler with command string",
    tool_call_limit=10
)

# Create task with handler
task = Task(
    description="""
    Create a 'customers' table with columns:
    - id (integer primary key)
    - name (text)
    - email (text)
    - phone (text)
    - registration_date (text)
    
    Insert 4 sample customers with different names and emails.
    Then query and show all customers.
    """,
    tools=[mcp_handler]
)

# Execute
result = agent.print_do(task)
print("Result:", result)
```

### Server Parameters

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler
from mcp.client.stdio import StdioServerParameters

# Create MCP handler with server parameters
mcp_handler = MCPHandler(
    server_params=StdioServerParameters(
        command="uvx",
        args=["mcp-server-sqlite", "--db-path", "/tmp/db.db"],
        env=None
    ),
    timeout_seconds=60
)

# Create agent
agent = Agent(
    name="MCPHandler Params Agent",
    role="Database operations specialist using MCPHandler with server params",
    goal="Demonstrate MCPHandler with StdioServerParameters",
    tool_call_limit=10
)

# Create task
task = Task(
    description="""
    Create an 'orders' table with columns:
    - id (integer primary key)
    - customer_id (integer)
    - product_name (text)
    - quantity (integer)
    - total_price (real)
    - order_date (text)
    
    Insert 3 sample orders with different products and quantities.
    Then query and show all orders sorted by total_price descending.
    """,
    tools=[mcp_handler]
)

# Execute
result = agent.print_do(task)
print("Result:", result)
```

## Parameters

* `command` (str): Command string to run MCP server
* `server_params`: Pre-configured server parameters (StdioServerParameters, SSEClientParams, or StreamableHTTPClientParams)
* `timeout_seconds` (int): Connection timeout in seconds (default: 5)
* `include_tools` (List\[str]): Optional list of tool names to include
* `exclude_tools` (List\[str]): Optional list of tool names to exclude
* `tool_name_prefix` (str): Optional prefix to add to all tool names from this handler (useful for preventing tool name collisions)

## Features

* **Automatic Tool Discovery**: Discovers all available tools from the server
* **Resource Management**: Automatically cleans up connections after task completion
* **Error Handling**: Graceful error handling with proper cleanup
* **Tool Filtering**: Include or exclude specific tools

## MCP Types

MCPHandler supports every MCP transport type. Below are real-world examples for each.

### Stdio: UVX Command

The simplest way to run a Python-based MCP server. Pass a command string and MCPHandler handles the rest.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

mcp_handler = MCPHandler(
    command="uvx mcp-server-fetch",
    timeout_seconds=30
)

agent = Agent(
    name="Web Fetcher",
    role="Web content retrieval specialist",
    goal="Fetch and summarize web content using MCP fetch tools",
    tool_call_limit=5
)

task = Task(
    description="Fetch the content of https://httpbin.org/get and summarize the JSON response.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

### Stdio: NPX Command

Run Node.js-based MCP servers with `npx`. This example uses the official filesystem server.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

mcp_handler = MCPHandler(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    timeout_seconds=30
)

agent = Agent(
    name="File Manager",
    role="File system operations specialist",
    goal="Manage files and directories",
    tool_call_limit=5
)

task = Task(
    description="List the allowed directories, then list the contents of the first allowed directory.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

### Stdio: StdioServerParameters

For full control over the subprocess, pass `StdioServerParameters` directly. Useful when you need custom environment variables.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler
from mcp.client.stdio import StdioServerParameters

mcp_handler = MCPHandler(
    server_params=StdioServerParameters(
        command="uvx",
        args=["mcp-server-fetch"],
        env=None
    ),
    timeout_seconds=30
)

agent = Agent(
    name="API Tester",
    role="API testing and validation specialist",
    goal="Test API endpoints and report results",
    tool_call_limit=5
)

task = Task(
    description="Fetch https://httpbin.org/headers and tell me what User-Agent header was sent.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

### SSE: URL Shorthand

Connect to a remote MCP server over Server-Sent Events. Pass the SSE endpoint URL and set `transport="sse"`.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

mcp_handler = MCPHandler(
    url="https://my-mcp-server.example.com/sse",
    transport="sse",
    timeout_seconds=30
)

agent = Agent(
    name="Remote SSE Agent",
    role="Remote tool execution via SSE",
    goal="Execute tools on the remote MCP server",
    tool_call_limit=5
)

task = Task(
    description="Use the available tools to complete the task.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

### SSE: SSEClientParams

For authenticated SSE connections with custom headers and timeout control.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler, SSEClientParams

mcp_handler = MCPHandler(
    server_params=SSEClientParams(
        url="https://my-mcp-server.example.com/sse",
        headers={"Authorization": "Bearer your-api-token"},
        timeout=10,
        sse_read_timeout=120
    ),
    timeout_seconds=30
)

agent = Agent(
    name="Auth SSE Agent",
    role="Authenticated remote tool execution",
    goal="Execute tools on the authenticated MCP server",
    tool_call_limit=5
)

task = Task(
    description="Use the available tools to complete the task.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

**SSEClientParams fields:**

| Field              | Type             | Default  | Description                            |
| ------------------ | ---------------- | -------- | -------------------------------------- |
| `url`              | `str`            | required | SSE endpoint URL                       |
| `headers`          | `Dict[str, Any]` | `None`   | Custom HTTP headers (e.g. auth tokens) |
| `timeout`          | `float`          | `5`      | Connection timeout in seconds          |
| `sse_read_timeout` | `float`          | `300`    | SSE stream read timeout in seconds     |

### Streamable HTTP: URL Shorthand

Connect to a remote MCP server using the Streamable HTTP transport. This is the newest MCP transport and is stateless by design.

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

mcp_handler = MCPHandler(
    url="https://my-mcp-server.example.com/mcp",
    transport="streamable-http",
    timeout_seconds=30
)

agent = Agent(
    name="HTTP Agent",
    role="Remote tool execution via Streamable HTTP",
    goal="Execute tools on the remote MCP server",
    tool_call_limit=5
)

task = Task(
    description="Use the available tools to complete the task.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

### Streamable HTTP: StreamableHTTPClientParams

For authenticated connections with custom headers, timeouts, and HTTP auth.

```python theme={null}
from datetime import timedelta
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler, StreamableHTTPClientParams

mcp_handler = MCPHandler(
    server_params=StreamableHTTPClientParams(
        url="https://my-mcp-server.example.com/mcp",
        headers={"Authorization": "Bearer your-api-token"},
        timeout=timedelta(seconds=30),
        sse_read_timeout=timedelta(seconds=120)
    ),
    timeout_seconds=30
)

agent = Agent(
    name="Auth HTTP Agent",
    role="Authenticated remote tool execution",
    goal="Execute tools on the authenticated MCP server",
    tool_call_limit=5
)

task = Task(
    description="Use the available tools to complete the task.",
    tools=[mcp_handler]
)

result = agent.print_do(task)
print("Result:", result)
```

**StreamableHTTPClientParams fields:**

| Field                | Type             | Default  | Description                            |
| -------------------- | ---------------- | -------- | -------------------------------------- |
| `url`                | `str`            | required | Streamable HTTP endpoint URL           |
| `headers`            | `Dict[str, Any]` | `None`   | Custom HTTP headers (e.g. auth tokens) |
| `timeout`            | `timedelta`      | `30s`    | Request timeout                        |
| `sse_read_timeout`   | `timedelta`      | `5min`   | SSE stream read timeout                |
| `terminate_on_close` | `bool`           | `None`   | Terminate server session on close      |
| `auth`               | `httpx.Auth`     | `None`   | httpx authentication handler           |

## Example with Structured Output

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler
from pydantic import BaseModel

class DatabaseReport(BaseModel):
    """Structured output for database operations."""
    tables_created: int
    records_inserted: int
    sample_data: str
    summary: str

# Create MCP handler
mcp_handler = MCPHandler(
    command="uvx mcp-server-sqlite --db-path /tmp/db.db",
    timeout_seconds=60
)

# Create agent
agent = Agent(
    name="Structured Output Agent",
    role="Database operations with structured reporting",
    goal="Demonstrate structured output with MCPHandler",
    tool_call_limit=10
)

# Create task with structured output
task = Task(
    description="""
    Create a 'inventory' table with columns:
    - id (integer primary key)
    - item_name (text)
    - category (text)
    - quantity (integer)
    - unit_price (real)
    
    Insert 6 sample inventory items across different categories.
    Then provide a structured report including:
    - Number of tables created
    - Number of records inserted
    - Sample data (first 3 items)
    - Summary of operations
    """,
    tools=[mcp_handler],
    response_format=DatabaseReport
)

# Execute
result = agent.print_do(task)
print("Tables Created:", result.tables_created)
print("Records Inserted:", result.records_inserted)
print("Sample Data:", result.sample_data)
print("Summary:", result.summary)
```

## Tool Name Prefix

When working with multiple MCP handlers that expose tools with the same names, use `tool_name_prefix` to prevent tool name collisions:

```python theme={null}
from upsonic import Agent, Task
from upsonic.tools.mcp import MCPHandler

# Create two handlers pointing to different databases
# Without prefix, tools like "create_table" would collide
users_db = MCPHandler(
    command="uvx mcp-server-sqlite --db-path /tmp/users.db",
    tool_name_prefix="users_db",  # Tools become: users_db_create_table, users_db_read_query, etc.
    timeout_seconds=60
)

products_db = MCPHandler(
    command="uvx mcp-server-sqlite --db-path /tmp/products.db",
    tool_name_prefix="products_db",  # Tools become: products_db_create_table, products_db_read_query, etc.
    timeout_seconds=60
)

# Create agent with both handlers
agent = Agent(
    name="Multi-DB Agent",
    role="Database operations specialist",
    goal="Work with multiple databases without tool name collisions",
    tools=[users_db, products_db],
    tool_call_limit=15
)

# Create task that uses both databases
task = Task(
    description="""
    IMPORTANT: You have access to two separate databases with prefixed tools.
    - Users database: Use tools prefixed with 'users_db_' (e.g., users_db_create_table)
    - Products database: Use tools prefixed with 'products_db_' (e.g., products_db_create_table)
    
    Step 1: In the users database, create a 'users' table with id and name columns.
    Step 2: In the products database, create a 'products' table with id and product_name columns.
    Step 3: Insert 2 sample records into each table.
    Step 4: Query both tables to verify the data.
    """
)

result = agent.print_do(task)
print("Result:", result)
```

The `tool_name_prefix` is especially useful when:

* Multiple MCP servers expose tools with identical names
* You need to clearly distinguish which server's tool to use
* Working with multiple instances of the same server type (e.g., multiple databases)

## When to Use

* Single MCP server connection
* Need more control over connection parameters
* Want automatic resource cleanup
* Need tool filtering capabilities
* Need to prevent tool name collisions with `tool_name_prefix`