How to properly invoke non-hosted MCP server in multi-agent workflow?

[Taka499]

Please read this first

• Have you read the docs?Agents SDK docs
• Have you searched for related issues? Others may have had similar requests

Question

As mentioned in the docs and #446, the non-hosted MCP server (Streamable HTTP / SSE / stdio) needs either async with to start connection, or manually connect with server.connect() and server.cleanup().

This requires BOTH agent declaration AND Runner.run to be wrapped inside async with or connect() & cleanup().

from pathlib import Path
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

current_dir = Path(__file__).parent
samples_dir = current_dir / "sample_files"

async with MCPServerStdio(
    name="Filesystem Server via npx",
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
) as server:
    agent = Agent(
        name="Assistant",
        instructions="Use the files in the sample directory to answer questions.",
        mcp_servers=[server],
    )
    result = await Runner.run(agent, "List the files available to you.")
    print(result.final_output)

However, this connection method is not ideal for a multi-agent workflow.

server = MCPServerStdio(...)

agent_fs = Agent(
    name="File System Agent",
    instructions="Use the files in the sample directory to answer questions.",
    mcp_servers=[server],
)

agent_echo = Agent(
    name="Echo Agent",
    instructions="Just repeat what the user says",
)

orchestrator = Agent(
    name="Orchestrator Agent",
    instructions=(
        "If the user asks for any repository question, then use file system agent to answer."
        "Otherwise, forward to echo agent."
    ),
    handoffs=[agent_echo, agent_fs],
)

asyncio.run(Runner.run(orchestrator))

It is fine with a simple multi-agent, because we can simply wrap the whole code with async with server:.

But what about in a complicated workflow? What if each agent is defined in a single .py file and imported in an orchestrator file? What if there are multiple MCP servers assigned to multiple agents, which is not a good option to connect to all the MCP servers at the very beginning of the workflow?

It is hard to write a simple structure to achieve the MCP server connection. I wonder if there's any good roundabout to solve this issue. I thought about using on_handoff, but this cannot close the connection.

[seratch]

For your use case, we recommend using server.connect() when setting up your agent that relies on the MCP server, and calling server.cleanup() when shutting down your app.

In other words, if you’d like to organize the multi-agent source files, you might consider creating a wrapper that provides two methods:

1. create an EchoAgent (internally starting its MCP server and attaching it to the agent).
2. clean up all internal states, including MCP server instances.

I know this may not be ideal for you, but I hope this was helpful. Also, whenever a better idea comes up, please feel free to share them with us!

[Taka499]

@seratch Thank you for your suggestion and idea! I tried and worked with a wrapper to McpServerStdio. This will still instantiate all the servers at the very beginning, but with a much cleaner approach:

# mcp_manager.py
from typing import List

from agents.mcp import MCPServerStdio


class MCPServerManager:
    """
    Manages the lifecycle of stdio MCP servers for multi-agent workflows.

    This wrapper addresses the challenge of managing MCP server connections
    across multiple agents defined in separate files. Instead of wrapping
    each agent declaration in async context managers, register servers once
    and use this manager to handle all connection/cleanup operations.

    Usage:
        # In mcp_server.py:
        mcp_manager = MCPServerManager()
        my_server = mcp_manager.register_server(MCPServerStdio(...))

        # In run.py:
        async with mcp_manager:
            result = await Runner.run(agent, message)
    """

    def __init__(self):
        self.servers: List[MCPServerStdio] = []

    def register_server(self, server: MCPServerStdio) -> MCPServerStdio:
        """
        Register an MCP server for lifecycle management.

        Args:
            server: The MCPServerStdio instance to manage

        Returns:
            The same server instance (for convenience)
        """
        if server not in self.servers:
            self.servers.append(server)
        return server

    async def __aenter__(self):
        """Connect all registered servers when entering async context."""
        await self.connect_all()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Cleanup all registered servers when exiting async context."""
        await self.cleanup_all()
        return False  # Don't suppress exceptions

    async def connect_all(self):
        """Explicitly connect all registered servers."""
        for server in self.servers:
            await server.connect()

    async def cleanup_all(self):
        """Explicitly cleanup all registered servers."""
        for server in self.servers:
            await server.cleanup()

Register mcp servers before attaching to the agent:

from .mcp_manager import MCPServerManager

# Global MCP server manager for lifecycle management
mcp_manager = MCPServerManager()

server = mcp_manager.register_server(McpServerStdio(...))

agent = Agent(
    ...
    mcp_servers=[server],
   ...
)

Then wrap the Runner.run() with the manager:

async with mcp_manager:
   response = await Runner.run(agent, prompt)