MCP Integration

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

# 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-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"
)

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

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

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

# 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")

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}")

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