MCP

The Model Context Protocol (MCP) is an open protocol that standardizes how applications provide context to Large Language Models (LLMs). AgentOpera implements MCP to enable seamless integration between AI models and various external tools, data sources, and services.

AgentOpera MCP Architecture

AgentOpera implements MCP through a client-server architecture:

1. MCP Server Params: Configuration parameters for connecting to MCP servers

2. MCP Session: Manages the connection between AgentOpera and MCP servers

3. MCP Tool Adapters: Wrap MCP tools to make them compatible with AgentOpera

The implementation supports two primary transport mechanisms:

• StdIO: For local command-line tools and processes

• SSE (Server-Sent Events): For HTTP-based remote services

Using MCP in AgentOpera

AgentOpera makes it easy to use MCP-compatible tools through its adapter system. There are two primary ways to use MCP in AgentOpera:

Using StdIO MCP Server

The StdIO adapter allows AgentOpera to communicate with MCP servers that run as local processes:

import asyncio
from agentopera.mcp import StdioServerParams, mcp_server_tools
from agentopera.chatflow.agents import AssistantAgent
from agentopera.models.openai import OpenAIChatCompletionClient
from agentopera.core import CancellationToken

async def main():
    # Configure file system access
    server_params = StdioServerParams(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"]
    )
    
    # Get all tools from the server
    tools = await mcp_server_tools(server_params)
    
    # Create an agent with MCP tools
    agent = AssistantAgent(
        name="file_manager",
        model_client=OpenAIChatCompletionClient(model="gpt-4"),
        tools=tools
    )
    
    # Use the agent with MCP tools
    await agent.run(
        task="Create a file called notes.txt with today's date as content",
        cancellation_token=CancellationToken()
    )

if __name__ == "__main__":
    asyncio.run(main())

Using SSE MCP Server

For connecting to remote MCP servers over HTTP with Server-Sent Events:

import asyncio
from agentopera.mcp import SseServerParams, mcp_server_tools
from agentopera.chatflow.agents import AssistantAgent
from agentopera.models.openai import OpenAIChatCompletionClient
from agentopera.core import CancellationToken

async def main():
    # Configure remote MCP service connection
    server_params = SseServerParams(
        url="https://api.example.com/mcp/sse",
        headers={"Authorization": "Bearer your-api-key"},
        timeout=30,  # Connection timeout in seconds
    )
    
    # Get all tools from the remote server
    tools = await mcp_server_tools(server_params)
    
    # Create an agent with remote MCP tools
    agent = AssistantAgent(
        name="api_assistant",
        model_client=OpenAIChatCompletionClient(model="gpt-4"),
        tools=tools
    )
    
    # Use the agent with MCP tools
    await agent.run(
        task="Fetch the latest weather data for New York",
        cancellation_token=CancellationToken()
    )

if __name__ == "__main__":
    asyncio.run(main())

Common MCP Servers

AgentOpera can integrate with a variety of MCP servers, including:

• Filesystem MCP Server: Access and manipulate files

• Database MCP Servers: Query databases like PostgreSQL, MongoDB

• Web MCP Servers: Fetch content from the web

• API MCP Servers: Interface with various APIs and services

• Development Tool Servers: Git, GitHub, and other development tools