Add multi-agent MCP system example with modular architecture #1864
+1,696
−0
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
added 2 commits
October 6, 2025 23:50
This commit introduces a comprehensive example demonstrating how to build scalable multi-agent systems using the Model Context Protocol (MCP). The architecture supports coordination between multiple MCP servers and specialized agents, designed to scale from a few agents to 100+ agents.
Key components:
1. **MCP Client Manager** (`mcp_client_manager.py`)
- Manages multiple MCP server connections with robust lifecycle management
- Implements double-shielded cleanup (anyio + asyncio) to prevent cancel scope conflicts during shutdown
- Supports concurrent cleanup with configurable timeouts
- Uses MCPServerStreamableHttpParams for proper connection configuration
2. **Agent Factory** (`agent_factory.py`)
- Factory pattern for creating and configuring specialized agents at scale
- Supports dynamic agent registration via AgentConfig
- Creates custom tool-agents using Runner.run_streamed() with advanced configuration
- Enables agent orchestration with hierarchical coordination
3. **MCP Servers** (`servers/`)
- Three example servers demonstrating different domains:
- Math server: mathematical operations (add, multiply, power)
- Text server: text manipulation (reverse, count words, uppercase)
- Data server: data processing (filter, sort, aggregate)
- Each server built with FastMCP using streamable-http transport
4. **Examples**
- `multiple_mcp_servers_example.py`: Complete demonstration of orchestrator coordinating specialized agents
- `adding_agents_example.py`: Shows how to dynamically add new agent types to the factory
5. **Documentation**
- `README.md`: Quick start guide and common use cases
- `ARCHITECTURE.md`: Detailed architecture guide for scaling patterns
- `servers/README.md`: Server setup and usage instructions
Technical highlights:
- Streaming responses at all levels using ResponseTextDeltaEvent
- Custom tool functions with @function_tool decorator wrapping Runner.run_streamed()
- Proper error handling for connection failures and graceful shutdown
- Logger suppression for harmless SDK cleanup messages
- Reusable MCP connections across multiple agents for efficiency
This architecture enables building complex multi-agent workflows with proper separation of concerns, robust error handling, and the ability to scale to enterprise-level agent systems.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Comment on lines
+121
to
+133
| async def _inner(): | ||
| # AnyIO shield prevents cancel scopes from interrupting cleanup() | ||
| with anyio.CancelScope(shield=True): | ||
| await server.cleanup() | ||
|
|
||
| if timeout is None: | ||
| # Don't allow outer-task cancellation to kill cleanup | ||
| with suppress(asyncio.CancelledError): | ||
| await asyncio.shield(_inner()) | ||
| else: | ||
| # Timebox *outside* the shields | ||
| with suppress(asyncio.TimeoutError, asyncio.CancelledError): | ||
| await asyncio.wait_for(asyncio.shield(_inner()), timeout=timeout) |
There was a problem hiding this comment.
The _shielded_cleanup helper wraps server.cleanup() in anyio.CancelScope(shield=True), but the examples that call this manager use asyncio.run and never enter an AnyIO-managed context. Instantiating CancelScope outside AnyIO raises RuntimeError: Cancel scopes can only be used from async code run by anyio, so the cleanup coroutine fails before server.cleanup() executes and the connection is left open. This contradicts the goal of robust shutdown. The shielded section should avoid AnyIO primitives or ensure cleanup is executed from an AnyIO task group.
Useful? React with 👍 / 👎.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This commit introduces a comprehensive example demonstrating how to build scalable multi-agent systems using the Model Context Protocol (MCP). The architecture supports coordination between multiple MCP servers and specialized agents, designed to scale from a few agents to 100+ agents.
Key components:
MCP Client Manager (
mcp_client_manager.py)Agent Factory (
agent_factory.py)MCP Servers (
servers/)Examples
multiple_mcp_servers_example.py: Complete demonstration of orchestrator coordinating specialized agentsadding_agents_example.py: Shows how to dynamically add new agent types to the factoryDocumentation
README.md: Quick start guide and common use casesARCHITECTURE.md: Detailed architecture guide for scaling patternsservers/README.md: Server setup and usage instructionsTechnical highlights:
This architecture enables building complex multi-agent workflows with proper separation of concerns, robust error handling, and the ability to scale to enterprise-level agent systems.