Skip to main content
The Model Context Protocol (MCP) provides a standard way to connect AI agents to external tools and data sources. Use MCP to give your agents access to filesystems, databases, APIs, and more.

​
What is MCP?

MCP servers expose tools that agents can use. Examples include:
  • Filesystem server: Read/write files
  • Database server: Query SQL databases
  • GitHub server: Manage repositories
  • Slack server: Send messages
  • Custom servers: Any tool you build

​
Basic Usage

import os
import asyncio
from opper_agents import Agent, mcp, MCPServerConfig

# Configure an MCP server
filesystem = MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()]
)

# Create agent with MCP tools
agent = Agent(
    name="FileAgent",
    description="Works with files",
    tools=[mcp(filesystem)]
)

async def main():
    # Agent can now read/write files
    result = await agent.process("List all Python files in the project")
    print(result)

asyncio.run(main())

​
Transport Types

​
stdio Transport

Runs the MCP server as a subprocess: 
# Using npx
server = MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "."]
)

# Using Docker
server = MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="docker",
    args=["run", "-i", "--rm", "mcp/filesystem", "/workspace"]
)

# Using a local binary
server = MCPServerConfig(
    name="myserver",
    transport="stdio",
    command="/usr/local/bin/my-mcp-server"
)

​
HTTP Transport

Connect to remote MCP servers over HTTP: 
# HTTP-SSE transport
server = MCPServerConfig(
    name="remote-api",
    transport="http-sse",
    url="https://api.example.com/mcp",
    headers={"Authorization": "Bearer token123"}
)

# Streamable HTTP transport
server = MCPServerConfig(
    name="remote-api",
    transport="streamable-http",
    url="https://api.example.com/mcp",
    headers={"Authorization": "Bearer token123"},
    session_id="optional-session-id"
)

​
Multiple MCP Servers

Use multiple servers together: 
import os
from opper_agents import Agent, mcp, MCPServerConfig

filesystem = MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()]
)

sqlite = MCPServerConfig(
    name="sqlite",
    transport="stdio",
    command="uvx",
    args=["mcp-server-sqlite", "--db-path", "./database.db"]
)

github = MCPServerConfig(
    name="github",
    transport="http-sse",
    url="https://mcp.github.example.com",
    headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}
)

agent = Agent(
    name="DevAgent",
    description="Development assistant with file, database, and GitHub access",
    tools=[
        mcp(filesystem),
        mcp(sqlite),
        mcp(github)
    ]
)

​
Tool Naming

MCP tools are automatically prefixed with the server name: 
filesystem:read_file
filesystem:write_file
filesystem:list_directory
sqlite:query
sqlite:execute
github:create_issue
github:list_repos
This prevents naming conflicts between servers.

​
Common MCP Servers

​
Filesystem

Access local files: 
filesystem = MCPServerConfig(
    name="fs",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/allowed/path"]
)

​
SQLite

Query SQLite databases: 
sqlite = MCPServerConfig(
    name="db",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-sqlite", "path/to/db.sqlite"]
)

​
Composio (Third-party Integrations)

Connect to services like Gmail, Slack, etc: 
# Gmail via Composio
gmail = MCPServerConfig(
    name="gmail",
    transport="http-sse",
    url="https://mcp.composio.dev/gmail",
    headers={"X-API-Key": COMPOSIO_API_KEY}
)

agent = Agent(
    name="EmailAgent",
    tools=[mcp(gmail)]
)

result = await agent.process("Find my unread emails from today")

​
Error Handling

MCP errors are handled like regular tool errors: 
from opper_agents import hook
from opper_agents.base.context import AgentContext

@hook("tool_error")
async def on_error(context: AgentContext, agent, tool, error, **kwargs):
    if tool.name.startswith("filesystem:"):
        print(f"File operation failed: {error}")
    elif tool.name.startswith("sqlite:"):
        print(f"Database error: {error}")

​
Lifecycle Management

MCP servers are automatically:
  • Started when the agent begins processing
  • Kept alive during execution
  • Cleaned up when the agent completes
For long-running agents, servers stay connected across iterations.

​
Example: File Listing Agent

import os
import asyncio
from opper_agents import Agent, mcp, MCPServerConfig

filesystem = MCPServerConfig(
    name="fs",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()]
)

agent = Agent(
    name="FileAgent",
    description="Lists files in a directory",
    tools=[mcp(filesystem)],
)

async def main():
    result = await agent.process("List the .py files in the current directory")
    print(result)

asyncio.run(main())

​
Best Practices

  1. Limit permissions: Only grant access to necessary directories/resources
  2. Use Docker for isolation: Run untrusted servers in containers
  3. Handle timeouts: MCP operations can be slow; set appropriate timeouts
  4. Monitor usage: Track which MCP tools are being called
  5. Secure credentials: Use environment variables for API keys

​
Next Steps

Tools

Combine MCP with custom tools

Examples

See MCP in action