MCP streamable HTTP Transport Help

[SivaShankar-Innocito]

Pre-submission Checklist

• I have checked that this question would not be more appropriate as an issue in a specific repository
• I have searched existing discussions and documentation for answers

Question Category

• Protocol Specification
• SDK Usage
• Server Implementation
• General Implementation
• Documentation
• Other

Your Question

I’m trying to connect multiple servers with a local client, but I keep getting errors and can’t figure out why. Can someone help me out?

client.py
main.py
ev_points_server.py
weather_server.py

[isheng-eqi]

I reviewed your code. There is one critical bug and several secondary issues preventing multi-server connections from working. Here is the diagnosis with fixes.

Critical bug: self.session is overwritten on every server connection

In client.py, connect_to_servers iterates through server configs and calls connect_to_streamable_http_server for each one. But every call overwrites self.session:

# current code — BROKEN
for config in server_configs:
    if await self.connect_to_streamable_http_server(server_name, server_url, headers):
        successful_connections += 1

Inside connect_to_streamable_http_server:

self.session: ClientSession = await self._session_context.__aenter__()

Only the last server's session survives. All previous sessions are lost because the same self.session variable is reassigned.

Later, your code tries to iterate self.session.items() (in _refresh_all_tools_cache) and access self.sessions[server_name] (in process_query), but self.session is a single ClientSession object, not a dict.

The fix

Replace the single self.session with a dict of sessions:

# In __init__:
self.sessions: Dict[str, ClientSession] = {}  # was: self.session = None
self._session_contexts: Dict[str, Any] = {}    # track per-server contexts
self._connection_managers: Dict[str, Any] = {} # track per-server transports

Then in connect_to_streamable_http_server, store sessions keyed by server name:

async def connect_to_streamable_http_server(self, server_name, server_url, headers=None):
    print(f"Connecting to {server_name} at {server_url}...")
    transport = streamablehttp_client(url=server_url, headers=headers or {})
    
    try:
        read_stream, write_stream, _ = await transport.__aenter__()
    except Exception as e:
        print(f"Failed to connect to {server_name}: {e}")
        await transport.__aexit__(None, None, None)
        return False
    
    try:
        session_ctx = ClientSession(read_stream, write_stream)
        session = await session_ctx.__aenter__()
        await session.initialize()
        
        # Store per-server references
        self.sessions[server_name] = session
        self._session_contexts[server_name] = session_ctx
        self._connection_managers[server_name] = transport
        
        print(f"Connected to {server_name}")
        return True
    except Exception as e:
        print(f"Session init failed for {server_name}: {e}")
        await transport.__aexit__(None, None, None)
        return False

Then fix _refresh_all_tools_cache:

async def _refresh_all_tools_cache(self):
    self._cached_tools = {}
    for server_name, session in self.sessions.items():  # was: self.session.items()
        try:
            response = await asyncio.wait_for(session.list_tools(), timeout=10.0)
            self._cached_tools[server_name] = response.tools
            print(f"Cached {len(response.tools)} tools from {server_name}")
        except Exception as e:
            print(f"Error caching tools for {server_name}: {e}")

And fix _convert_tools_for_anthropic to aggregate across all servers:

async def _convert_tools_for_anthropic(self) -> List[dict]:
    all_tools = []
    for server_name, session in self.sessions.items():
        response = await session.list_tools()
        for tool in response.tools:
            all_tools.append({
                "name": tool.name,
                "description": f"[{server_name}] {tool.description}",
                "input_schema": tool.inputSchema,
            })
    return all_tools

Secondary issue: cleanup needs to handle all servers

Your current cleanup only handles one session. Fix:

async def cleanup(self):
    for server_name in list(self._session_contexts.keys()):
        try:
            await self._session_contexts[server_name].__aexit__(None, None, None)
        except Exception as e:
            print(f"Cleanup error for {server_name}: {e}")
    for server_name in list(self._connection_managers.keys()):
        try:
            await self._connection_managers[server_name].__aexit__(None, None, None)
        except Exception as e:
            print(f"Transport cleanup error for {server_name}: {e}")
    self.sessions.clear()

Root cause summary

You designed MultiServerMCPClient with the right intent (a dict of sessions, keyed by server name), but the storage variable is a single self.session instead of self.sessions dict. Every new connection silently overwrites the previous one, so only the last server actually works. The fix is mechanical: use a dict everywhere.

[SivaShankar-Innocito]

Thank you for the review and the answer helps me a lot.