MCP Integration¶

from selectools import Agent, AgentConfig
from selectools.providers.stubs import LocalProvider
from selectools.mcp import mcp_tools, MCPServerConfig

# Connect to an MCP server and use its tools
with mcp_tools(MCPServerConfig(command="python", args=["server.py"])) as tools:
    agent = Agent(
        tools=tools,
        provider=LocalProvider(),
        config=AgentConfig(model="gpt-4o-mini"),
    )
    result = agent.run("Search for Python tutorials")
    print(result.content)

from selectools import Agent, AgentConfig
from selectools.providers import OpenAIProvider
from selectools.mcp import mcp_tools, MCPServerConfig

# Connect to an MCP server and get tools
with mcp_tools(MCPServerConfig(command="python", args=["server.py"])) as tools:
    agent = Agent(
        provider=OpenAIProvider(),
        config=AgentConfig(model="gpt-4.1-mini"),
        tools=tools,
    )
    result = agent.run("Search for Python tutorials")

from selectools.mcp import MCPServerConfig

# stdio transport (local subprocess)
config = MCPServerConfig(
    command="python",
    args=["my_server.py"],
    name="search",
)

# Streamable HTTP transport (remote)
config = MCPServerConfig(
    url="http://api.example.com/mcp",
    transport="streamable-http",
    headers={"Authorization": "Bearer token"},
    name="api",
)

from selectools.mcp import MCPClient, MCPServerConfig

# Async context manager (preferred)
async with MCPClient(config) as client:
    tools = await client.list_tools()
    result = await client._call_tool("search", {"query": "python"})

# Sync context manager
with MCPClient(config) as client:
    tools = client.list_tools_sync()

config = MCPServerConfig(
    command="python",
    args=["unreliable_server.py"],
    circuit_breaker_threshold=3,     # Open after 3 failures
    circuit_breaker_cooldown=60.0,   # Retry after 60 seconds
)

config = MCPServerConfig(
    command="python",
    args=["server.py"],
    max_retries=3,          # Retry up to 3 times
    retry_backoff=1.0,      # 1s, 2s, 4s exponential backoff
)

from selectools.mcp import MultiMCPClient, MCPServerConfig

async with MultiMCPClient([
    MCPServerConfig(command="python", args=["search.py"], name="search"),
    MCPServerConfig(url="http://api.example.com/mcp",
                    transport="streamable-http", name="api"),
]) as client:
    tools = await client.list_all_tools()
    # Tools are prefixed: search_web_search, api_query, etc.
    agent = Agent(provider=p, tools=tools, config=c)

async with MultiMCPClient(configs) as client:
    print(f"Active: {client.active_servers}")   # ["search"]
    print(f"Failed: {client.failed_servers}")   # ["api"]
    tools = await client.list_all_tools()       # Only search tools

MultiMCPClient(configs, prefix_tools=True)   # search_query, api_fetch
MultiMCPClient(configs, prefix_tools=False)  # Raises ValueError on collision

from selectools import tool
from selectools.mcp import MCPServer

@tool(description="Get weather for a city")
def get_weather(city: str) -> str:
    return f"72°F in {city}"

@tool(description="Search documents")
def search(query: str) -> str:
    return f"Results for: {query}"

server = MCPServer(tools=[get_weather, search])
server.serve(transport="stdio")
# or: server.serve(transport="streamable-http", port=8080)

import asyncio
from selectools import Agent, AgentConfig, tool
from selectools.providers import AnthropicProvider
from selectools.mcp import MCPClient, MCPServerConfig

@tool(description="Local calculator")
def multiply(a: int, b: int) -> str:
    return str(a * b)

async def main():
    config = MCPServerConfig(command="python", args=["math_server.py"])

    async with MCPClient(config) as client:
        mcp_tools = await client.list_tools()

        # Mix local + MCP tools
        agent = Agent(
            provider=AnthropicProvider(),
            config=AgentConfig(model="claude-haiku-4-5"),
            tools=[multiply] + mcp_tools,
        )

        # Agent automatically selects the right tool
        result = await agent.arun("Add 5 and 3")      # Uses MCP 'add' tool
        result2 = await agent.arun("Multiply 6 by 7")  # Uses local 'multiply'

asyncio.run(main())

from selectools.evals import EvalSuite, TestCase

suite = EvalSuite(
    agent=agent,  # Agent with MCP tools
    cases=[
        TestCase(input="Add 10 and 20", expect_tool="add"),
        TestCase(input="Search for Python", expect_tool="search"),
    ],
)
report = suite.run()
print(report.accuracy)