tessl/pypi-openai-agents

Lightweight framework for building multi-agent workflows with LLMs, supporting handoffs, guardrails, tools, and 100+ LLM providers

Overview

Eval results

Files

Model Context Protocol (MCP)

Name: tessl/pypi-openai-agents
Author: tessl

The Model Context Protocol (MCP) enables agents to connect to external servers that provide tools, resources, and capabilities beyond the base SDK. MCP servers can be accessed via stdio, SSE, or streamable HTTP transports, with support for tool filtering and approval workflows.

Capabilities

MCP Server Types

Union type for all MCP server implementations.

MCPServer = Union[
    MCPServerStdio,
    MCPServerSse,
    MCPServerStreamableHttp
]

MCP Server Stdio

MCP server connection via standard input/output.

class MCPServerStdio:
    """MCP server via stdio transport."""

    def __init__(params: MCPServerStdioParams):
        """
        Initialize stdio MCP server.

        Parameters:
        - params: Stdio connection parameters
        """

    async def connect():
        """Establish connection to MCP server."""

    async def cleanup():
        """Clean up server connection."""

class MCPServerStdioParams:
    """
    Parameters for stdio MCP server.

    Attributes:
    - command: str - Command to execute
    - args: list[str] - Command arguments
    - env: dict[str, str] - Environment variables
    """

Usage example:

from agents import Agent
from agents.mcp import MCPServerStdio

# Git MCP server via stdio
git_server = MCPServerStdio({
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-git"],
    "env": {}
})

agent = Agent(
    name="Git Agent",
    instructions="Help with git operations",
    mcp_servers=[git_server]
)

MCP Server SSE

MCP server connection via Server-Sent Events.

class MCPServerSse:
    """MCP server via SSE transport."""

    def __init__(params: MCPServerSseParams):
        """
        Initialize SSE MCP server.

        Parameters:
        - params: SSE connection parameters
        """

    async def connect():
        """Establish connection to MCP server."""

    async def cleanup():
        """Clean up server connection."""

class MCPServerSseParams:
    """
    Parameters for SSE MCP server.

    Attributes:
    - url: str - SSE endpoint URL
    - headers: dict[str, str] - HTTP headers
    """

Usage example:

from agents.mcp import MCPServerSse

# Remote MCP server via SSE
sse_server = MCPServerSse({
    "url": "https://example.com/mcp-sse",
    "headers": {"Authorization": "Bearer token"}
})

agent = Agent(
    name="Remote Agent",
    mcp_servers=[sse_server]
)

MCP Server Streamable HTTP

MCP server connection via streamable HTTP.

class MCPServerStreamableHttp:
    """MCP server via streamable HTTP transport."""

    def __init__(params: MCPServerStreamableHttpParams):
        """
        Initialize streamable HTTP MCP server.

        Parameters:
        - params: HTTP connection parameters
        """

    async def connect():
        """Establish connection to MCP server."""

    async def cleanup():
        """Clean up server connection."""

class MCPServerStreamableHttpParams:
    """
    Parameters for streamable HTTP MCP server.

    Attributes:
    - url: str - HTTP endpoint URL
    - headers: dict[str, str] - HTTP headers
    """

Usage example:

from agents.mcp import MCPServerStreamableHttp

http_server = MCPServerStreamableHttp({
    "url": "https://example.com/mcp-http",
    "headers": {"X-API-Key": "key"}
})

agent = Agent(
    name="HTTP Agent",
    mcp_servers=[http_server]
)

MCP Utilities

Utility functions for working with MCP.

class MCPUtil:
    """MCP utilities."""

    @staticmethod
    async def get_all_function_tools(
        servers: list[MCPServer],
        tool_filter: ToolFilter | None = None,
        context = None
    ) -> list[Tool]:
        """
        Get all MCP tools from servers.

        Parameters:
        - servers: List of MCP servers
        - tool_filter: Optional filter for tools
        - context: Context object

        Returns:
        - list[Tool]: All tools from MCP servers
        """

Tool Filtering

Filter which MCP tools are available to agents.

ToolFilter = Union[ToolFilterStatic, ToolFilterCallable]

class ToolFilterStatic:
    """
    Static tool filter.

    Attributes:
    - allowed_tools: list[str] | None - Allowed tool names (whitelist)
    - denied_tools: list[str] | None - Denied tool names (blacklist)
    """

ToolFilterCallable = Callable[[ToolFilterContext], bool]

class ToolFilterContext:
    """
    Context for tool filtering.

    Attributes:
    - tool_name: str - Name of tool being filtered
    - server_label: str - Label of MCP server
    """

def create_static_tool_filter(
    allowed_tools: list[str] | None = None,
    denied_tools: list[str] | None = None
) -> ToolFilterStatic:
    """
    Create static tool filter.

    Parameters:
    - allowed_tools: Whitelist of tool names
    - denied_tools: Blacklist of tool names

    Returns:
    - ToolFilterStatic: Configured filter
    """

Usage example:

from agents.mcp import create_static_tool_filter

# Whitelist specific tools
filter = create_static_tool_filter(
    allowed_tools=["read_file", "list_directory"]
)

# Blacklist dangerous tools
filter = create_static_tool_filter(
    denied_tools=["delete_file", "execute_command"]
)

# Custom filter function
def custom_filter(ctx):
    """Only allow read operations."""
    return ctx.tool_name.startswith("read_") or ctx.tool_name.startswith("list_")

# Apply filter to agent
agent = Agent(
    name="Safe Agent",
    mcp_servers=[mcp_server],
    # Filter applied via MCPConfig or tool_filter parameter
)

MCP Configuration

Configure MCP behavior per agent.

from agents import Agent, MCPConfig

config = MCPConfig(
    convert_schemas_to_strict=True  # Convert MCP tool schemas to strict mode
)

agent = Agent(
    name="MCP Agent",
    mcp_servers=[filesystem_server, git_server],
    mcp_config=config
)

Common MCP Servers

Filesystem MCP Server

Access local filesystem.

from agents.mcp import MCPServerStdio

filesystem_server = MCPServerStdio({
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/workspace"],
    "env": {}
})

agent = Agent(
    name="File Agent",
    instructions="Help with file operations",
    mcp_servers=[filesystem_server]
)

Git MCP Server

Git repository operations.

git_server = MCPServerStdio({
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-git"],
    "env": {"GIT_REPO_PATH": "/path/to/repo"}
})

agent = Agent(
    name="Git Agent",
    instructions="Help with git operations",
    mcp_servers=[git_server]
)

Custom MCP Server

Create your own MCP server in Python:

# server.py - Simple MCP server
from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("my-custom-server")

@server.tool("get_weather")
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny"

@server.tool("calculate")
def calculate(expression: str) -> float:
    """Evaluate math expression."""
    return eval(expression)

if __name__ == "__main__":
    stdio_server(server)

Use custom server:

custom_server = MCPServerStdio({
    "command": "python",
    "args": ["server.py"],
    "env": {}
})

agent = Agent(
    name="Custom Agent",
    mcp_servers=[custom_server]
)

Multiple MCP Servers

Use multiple MCP servers together:

filesystem_server = MCPServerStdio({
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]
})

git_server = MCPServerStdio({
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-git"]
})

web_server = MCPServerSse({
    "url": "https://mcp-server.example.com/sse",
    "headers": {}
})

agent = Agent(
    name="Multi-Tool Agent",
    instructions="Use filesystem, git, and web tools as needed",
    mcp_servers=[filesystem_server, git_server, web_server]
)

Tool Approval Workflow

Require approval for sensitive MCP tool calls:

from agents import HostedMCPTool

async def approve_mcp_action(request):
    """Review and approve MCP tool calls."""
    action = request.data.get("action")
    tool = request.data.get("tool")

    print(f"MCP tool approval request:")
    print(f"  Tool: {tool}")
    print(f"  Action: {action}")

    # Check if action is safe
    if "delete" in action.lower() or "rm" in action.lower():
        return {
            "approve": False,
            "reason": "Destructive operations require manual approval"
        }

    return {"approve": True}

# Use with hosted MCP tool
hosted_mcp = HostedMCPTool(
    tool_config={"server": "mcp_server_id"},
    on_approval_request=approve_mcp_action
)

agent = Agent(
    name="Safe Agent",
    tools=[hosted_mcp]
)

Best Practices

Security: Use tool filters to restrict dangerous operations
Server Lifecycle: Properly cleanup MCP servers after use
Error Handling: Handle MCP server connection failures gracefully
Tool Discovery: Let agents discover available MCP tools dynamically
Approval Workflows: Require approval for sensitive operations
Server Selection: Choose appropriate transport (stdio, SSE, HTTP) for use case
Environment Variables: Use env params for server configuration
Multiple Servers: Combine servers for comprehensive tool coverage
Testing: Test MCP integrations thoroughly before production
Documentation: Document available MCP tools for agent instructions