python-sdk
python-sdk copied to clipboard
feat: add MCP proxy pattern convenience function (#12)
Description
Implements the mcp_proxy() convenience function that enables bidirectional message forwarding between two MCP transports, porting the TypeScript proxy pattern to the Python SDK.
Changes
- Added
mcp_proxy()context manager insrc/mcp/shared/proxy.py - Implements bidirectional message forwarding using
anyiotask groups - Supports optional error handling with sync/async callbacks
- Automatically closes both transports when one closes
- Comprehensive test coverage in
tests/shared/test_proxy.py
Features
- Bidirectional forwarding: Messages flow in both directions simultaneously
- Error handling: Optional
onerrorcallback (supports both sync and async) - Automatic cleanup: When one transport closes, the other is also closed
- Exception handling: Properly handles
SessionMessageandExceptionobjects from read streams - Resource management: Uses context manager pattern for proper cleanup
Usage Example
from mcp.shared.proxy import mcp_proxy
async with mcp_proxy( transport_to_client=(client_read, client_write), transport_to_server=(server_read, server_write), onerror=lambda e: logger.error(f"Proxy error: {e}"), ): # Proxy is active, forwarding messages bidirectionally await some_operation()
Both transports are closed when exiting the context## Testing
- Unit tests cover bidirectional forwarding, error handling, stream cleanup, and edge cases
- All tests pass with proper async/await patterns
- Follows existing test patterns in the repository
Related
Closes #12
Type of Change
- [x] New feature (non-breaking change which adds functionality)
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Documentation update
Checklist
- [x] Code follows the repository's style guidelines
- [x] Tests added/updated for new functionality
- [x] All tests pass locally
- [x] Linting and type checking pass
- [x] Documentation updated (if applicable)