Skip to content

Conversation

@dgenio
Copy link

@dgenio dgenio commented Nov 28, 2025

Summary

Implements the MCP proxy pattern to simplify bidirectional message forwarding between two transports, resolving #12.

This PR adds a convenience function mcp_proxy() that enables proxy/gateway use cases by transparently forwarding MCP messages between client and server transports with optional inspection and transformation capabilities.

Changes

Core Implementation

  • src/mcp/shared/proxy.py: New module with mcp_proxy() async context manager
    • Bidirectional message forwarding (client ↔ server)
    • Optional message inspection/transformation via callbacks
    • Robust error handling with continuation (errors don't stop the proxy)
    • Automatic stream cleanup on context exit
    • Full type hints and comprehensive docstrings

API Export

  • src/mcp/__init__.py: Added mcp_proxy to public API exports

Testing

  • tests/shared/test_proxy.py: Comprehensive test suite with 12 test cases covering:
    • Bidirectional message forwarding
    • Message transformation and filtering
    • Error handling and recovery
    • Stream cleanup
    • Multiple message handling

Example

  • examples/servers/simple-proxy/: Complete working example demonstrating:
    • Transparent proxy server implementation
    • Message inspection/logging
    • CLI interface with Click
    • Production-ready error handling
    • Comprehensive README with usage examples

Key Features

  1. Stream-based Design: Adapts the TypeScript event-based pattern to Python's async stream model using anyio and memory streams
  2. Optional Transformation: Callbacks can inspect, modify, or drop messages before forwarding
  3. Error Resilience: Errors in one message don't stop forwarding of other messages
  4. Clean Resource Management: Automatic cleanup of all streams via context manager
  5. Type-Safe: Full type hints compatible with Pyright

Example Usage

from mcp.shared.proxy import mcp_proxy
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp.server.stdio import stdio_server

async def log_messages(msg: SessionMessage) -> SessionMessage:
    print(f"Forwarding: {msg.message.root.method}")
    return msg

async with stdio_client(server_params) as server_streams, \
           stdio_server() as client_streams:
    async with mcp_proxy(
        client_streams=client_streams,
        server_streams=server_streams,
        onerror=lambda e: print(f"Error: {e}"),
        on_client_message=log_messages,
    ):
        await anyio.sleep_forever()## Use Cases
  • Debugging: Inspect MCP protocol messages in real-time
  • Monitoring: Log and track MCP interactions
  • Gateway: Add authentication, rate limiting, or caching layers
  • Testing: Inject test messages or simulate network conditions

Implementation Notes

This implementation differs from the TypeScript version by:

  • Using async streams instead of event emitters
  • Using context managers for resource cleanup instead of explicit close handlers
  • Following Python's async/await patterns with anyio

Testing

All tests pass:

  • ✅ 12 test cases for proxy functionality
  • ✅ No linter errors (ruff)
  • ✅ Type checking passes (pyright)
  • ✅ Follows repository code style

Documentation

  • Comprehensive docstrings with examples
  • Working example server in examples/servers/simple-proxy/
  • Detailed README with architecture diagram and usage examples

Closes #12

@dgenio dgenio closed this Nov 28, 2025
@dgenio dgenio force-pushed the feature/mcp-proxy-pattern branch from a9aa0bf to 5983a65 Compare November 28, 2025 17:58
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add conveniences for MCP proxy pattern

1 participant