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

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

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.

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.

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.

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".

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.

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.

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.

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

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:

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