MCP Tools Integration

Integrate Model Context Protocol tools with agents

MCP Tools Integration

The MCP (Model Context Protocol) module enables seamless integration of MCP-compliant tools and servers with PraisonAI agents, supporting multiple transport methods.

Diagram

Overview

MCP (Model Context Protocol) is a standard for connecting AI assistants to external tools and data sources. The MCP module in PraisonAI provides:

stdio Transport: Run MCP servers as subprocess commands
Streamable HTTP Transport: Connect to HTTP endpoints with session management
WebSocket Transport: Bidirectional real-time connections (SEP-1288)
SSE Transport (Legacy): Backward compatibility with older servers
Automatic Tool Discovery: Tools are automatically discovered and made available to agents
Session Management: Stateful sessions with Mcp-Session-Id header
Resumability: SSE stream recovery via Last-Event-ID
Security: Origin validation, authentication headers, secure session IDs

Quick Start

NPX MCP Server

from praisonaiagents import Agent, MCP

# Use the memory MCP server from NPX
agent = Agent(
    name="Memory Assistant",
    instructions="You can store and retrieve memories.",
    tools=MCP("npx @modelcontextprotocol/server-memory")
)

response = agent.start("Remember that my favourite colour is blue")

Python MCP Server

# Python MCP server
agent = Agent(
    name="Data Assistant",
    tools=MCP("/usr/bin/python3 /path/to/mcp_server.py")
)

# Or with separate command and args
agent = Agent(
    tools=MCP(
        command="/usr/bin/python3",
        args=["/path/to/mcp_server.py", "--config", "config.json"]
    )
)

HTTP Endpoint

# Streamable HTTP endpoint (auto-detected)
agent = Agent(
    name="Remote Assistant",
    tools=MCP("https://api.example.com/mcp")
)

WebSocket

# WebSocket endpoint (auto-detected via ws://)
agent = Agent(
    name="Realtime Assistant",
    tools=MCP("wss://api.example.com/mcp")
)

SSE Legacy

# Legacy SSE endpoint (auto-detected via /sse suffix)
agent = Agent(
    tools=MCP("http://localhost:8080/sse")
)

Transport Methods

Stdio Transport

The stdio transport runs MCP servers as subprocesses:

Command String

# Simple command string
tools = MCP("npx @modelcontextprotocol/server-github")

# Python script
tools = MCP("python3 mcp_weather_server.py")

# With environment variables
import os
os.environ["API_KEY"] = "your-api-key"
tools = MCP("node weather-server.js")

Command and Args

# Separate command and arguments
tools = MCP(
    command="/usr/bin/python3",
    args=["server.py", "--port", "8080", "--debug"]
)

# Node.js server with args
tools = MCP(
    command="node",
    args=["dist/server.js", "--config", "production.json"]
)

Advanced Options

# With timeout and debug
tools = MCP(
    "npx @modelcontextprotocol/server-filesystem",
    timeout=120,  # 2 minutes timeout
    debug=True    # Enable debug logging
)

Streamable HTTP Transport

The Streamable HTTP transport (Protocol Revision 2025-11-25) provides session management and resumability:

Basic HTTP

# Streamable HTTP endpoint
tools = MCP("https://api.example.com/mcp")

With Session Management

# Enable session management
tools = MCP(
    "https://api.example.com/mcp",
    session=True,
    resumability=True
)

With Authentication

# HTTP with auth headers
tools = MCP(
    "https://api.example.com/mcp",
    headers={"Authorization": "Bearer token"}
)

WebSocket Transport

The WebSocket transport provides bidirectional real-time connections:

Basic WebSocket

# WebSocket endpoint (auto-detected)
tools = MCP("ws://localhost:8080/mcp")

# Secure WebSocket
tools = MCP("wss://api.example.com/mcp")

With Authentication

# WebSocket with auth token
tools = MCP(
    "wss://api.example.com/mcp",
    auth_token="Bearer your-token"
)

With Options

# With timeout and debug
tools = MCP(
    "wss://api.example.com/mcp",
    timeout=60,
    debug=True
)

SSE Transport (Legacy)

The SSE transport is maintained for backward compatibility with older servers:

Legacy SSE

# URLs ending in /sse use legacy transport
tools = MCP("http://localhost:8080/sse")

With Options

# With timeout
tools = MCP(
    "http://localhost:3000/agents/sse",
    timeout=30,
    debug=True
)

Available MCP Servers

Official NPX Servers

Card

bash

npx @modelcontextprotocol/server-memory

Store and retrieve conversation memories

Card

bash

npx @modelcontextprotocol/server-filesystem

Read and write files with safety controls

Card

bash

npx @modelcontextprotocol/server-github

Interact with GitHub repositories

Card

bash

npx @modelcontextprotocol/server-postgres

Query and manage PostgreSQL databases

Custom Python Servers

Create your own MCP server in Python:

python

# mcp_calculator.py
import asyncio
from mcp import Server, Tool

server = Server("calculator")

@server.tool
async def add(a: float, b: float) -> float:
    """Add two numbers"""
    return a + b

@server.tool
async def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    return a * b

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

Use in agent:

python

agent = Agent(
    tools=MCP("python3 mcp_calculator.py")
)

Tool Discovery

MCP automatically discovers available tools:

python

# Create MCP instance
mcp = MCP("npx @modelcontextprotocol/server-filesystem")

# Tools are automatically discovered and can be iterated
for tool in mcp:
    print(f"Tool: {tool.name}")
    print(f"Description: {tool.description}")
    print(f"Parameters: {tool.parameters}")

# Assign to agent
agent = Agent(name="File Manager", tools=mcp)

Error Handling

MCP includes built-in error handling and retry logic for robust operation.

python

try:
    agent = Agent(
        tools=MCP("npx @modelcontextprotocol/server-memory")
    )
    response = agent.start("Store this information")
except Exception as e:
    print(f"MCP Error: {e}")
    # Fallback logic

Advanced Usage

Multiple MCP Servers

PraisonAI supports passing multiple MCP server instances directly to an agent. You can combine different MCP servers along with regular Python functions.

Direct List (Recommended)

from praisonaiagents import Agent, MCP

# Pass multiple MCP instances directly in a list
agent = Agent(
    name="Multi-Tool Assistant",
    instructions="You can manage files, memories, and check time.",
    tools=[
        MCP("uvx mcp-server-time"),                     # Time tools
        MCP("npx @modelcontextprotocol/server-memory"), # Memory tools
        MCP("npx @modelcontextprotocol/server-filesystem /tmp")  # File tools
    ]
)

response = agent.start("What time is it in Tokyo? Then save this to memory.")

Mixed MCP + Functions

from praisonaiagents import Agent, MCP

def my_custom_tool(query: str) -> str:
    """A custom Python function tool."""
    return f"Custom result for: {query}"

# Combine MCP servers with regular Python functions
agent = Agent(
    name="Hybrid Assistant",
    instructions="You have access to time tools and a custom search.",
    tools=[
        MCP("uvx mcp-server-time"),  # MCP server
        my_custom_tool                # Regular function
    ]
)

Spread Operator (Legacy)

from praisonaiagents import Agent, MCP

# Alternative: spread MCP tools into a list
memory_tools = MCP("npx @modelcontextprotocol/server-memory")
file_tools = MCP("npx @modelcontextprotocol/server-filesystem")

agent = Agent(
    name="Multi-Tool Assistant",
    tools=[*memory_tools, *file_tools]
)

Environment Configuration

python

import os

# Set environment variables for MCP servers
os.environ["GITHUB_TOKEN"] = "your-github-token"
os.environ["DATABASE_URL"] = "postgresql://..."

# MCP servers can access these variables
github_tools = MCP("npx @modelcontextprotocol/server-github")
db_tools = MCP("python3 database_mcp.py")

Debugging MCP Connections

python

# Enable debug mode
mcp = MCP(
    "npx @modelcontextprotocol/server-memory",
    debug=True  # Prints detailed logs
)

# Check if tools are loaded
if not list(mcp):
    print("No tools discovered from MCP server")

Creating SSE MCP Servers

Example SSE server implementation:

python

# sse_mcp_server.py
from flask import Flask, Response
import json

app = Flask(__name__)

@app.route('/sse')
def sse():
    def generate():
        # Send tool definitions
        tools = [{
            "name": "get_weather",
            "description": "Get weather information",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }]

        yield f"data: {json.dumps({'type': 'tools', 'tools': tools})}\n\n"

    return Response(generate(), mimetype="text/event-stream")

if __name__ == '__main__':
    app.run(port=8080)

Best Practices

Transport Choice

Use stdio for local tools and development
Use SSE for remote/cloud deployments
Consider latency and reliability needs

Reliability

Implement timeouts for long-running operations
Handle server disconnections gracefully
Provide fallback options

Security

Validate input/output from MCP servers
Use environment variables for secrets
Implement proper authentication for SSE

Performance

Reuse MCP instances when possible
Monitor subprocess resource usage
Implement connection pooling for SSE

Example: Multi-Tool Agent

Complete Example

from praisonaiagents import Agent, Task, AgentTeam, MCP
import os

# Set up environment
os.environ["GITHUB_TOKEN"] = "ghp_..."

# Create agent with multiple MCP tools
agent = Agent(
    name="DevOps Assistant",
    instructions="""You are a DevOps assistant that can:
    - Manage files and directories
    - Interact with GitHub repositories
    - Store and retrieve important information
    Use your tools wisely to help with development tasks.""",
    tools=[
        MCP("npx @modelcontextprotocol/server-filesystem"),
        MCP("npx @modelcontextprotocol/server-github"),
        MCP("npx @modelcontextprotocol/server-memory")
    ]
)

# Create tasks
tasks = [
    Task(
        description="Check the current directory structure",
        agent=agent,
        expected_output="Directory listing with key files identified"
    ),
    Task(
        description="Remember the project structure for future reference",
        agent=agent,
        expected_output="Confirmation that structure is memorised"
    )
]

# Run the system
agents = AgentTeam(agents=[agent], tasks=tasks)
result = agents.start()

Resources & Prompts

If the connected server advertises resources or prompts capabilities, PraisonAI automatically registers agent-callable tools (list_mcp_resources, read_mcp_resource, list_mcp_prompts, get_mcp_prompt). Tools-only servers are unaffected — no extra tools appear.

See MCP Resources & Prompts for the full guide.

Next Steps

Card

Create custom tools for agents

Card

Explore available MCP servers