feat: add ClaudeSDKClient.set_mcp_servers()

[omgupta-duplo]

Summary

Adds ClaudeSDKClient.set_mcp_servers(servers), exposing the CLI's mcp_set_servers control request so a running session can change its MCP server configuration without restarting.

The motivating case is an OAuth-authenticated HTTP MCP server whose bearer token expires mid-session. .mcp.json and the subprocess environment are both fixed when the CLI is spawned, so a freshly minted token has no way to reach an already-connected server — today the only options are to end the turn or restart the client. reconnect_mcp_server re-reads the config file, but that requires writing the literal bearer to disk, which is not an option when the session directory is on shared storage. mcp_set_servers carries the new headers in-memory, over the control channel.

The part worth reviewing

mcp_set_servers is authoritative: it replaces the whole dynamically-managed server set. The in-process SDK servers registered from ClaudeAgentOptions.mcp_servers are part of that set, so a caller who pushes only their external server silently loses every SDK tool for the rest of the session — the CLI's own response reports them under removed.

Rather than make that a documented footgun, Query.set_mcp_servers pre-seeds the payload with the already-registered SDK servers, serialized instance-stripped as {"type": "sdk", "name": ...} — the same shape subprocess_cli sends at startup. So:

• SDK servers survive a push without the caller re-listing them.
• An entry carrying an instance is registered with an SdkMcpBridge, so a server added mid-session routes its tool calls back in-process; the instance is stripped before the config reaches the CLI, exactly as at startup.
• Caller entries take precedence on a name collision, so an SDK server can still be replaced deliberately.

Happy to change the preserve-by-default behavior to an explicit opt-in flag if you'd rather the method stay a thin pass-through — the trade-off is that the authoritative-replace semantics then have to be learned from a failure.

Test plan

Four tests in tests/test_streaming_client.py, mirroring the existing reconnect_mcp_server / toggle_mcp_server control-request tests:

• the control request is sent with the expected subtype and servers payload
• a registered SDK server is re-included in the payload alongside the pushed HTTP server
• an SDK server passed in is registered in sdk_mcp_servers and its instance does not reach the wire
• calling before connect() raises CLIConnectionError

python -m pytest tests/              # 1595 passed, 6 skipped
python -m ruff check src/ tests/ scripts/    # All checks passed
python -m ruff format --check ...            # 70 files already formatted
python -m mypy src/ scripts/                 # Success: no issues found in 33 source files

The underlying control request was also verified against a live CLI (2.1.251): a push naming only the external server returns {"added": ["carbonarc"], "removed": ["agent_tools"]} and the SDK tool stops existing, while re-including the SDK entry returns removed: [] and the tool keeps working.

[sigley]

On exact head 24c3d5b6730c10f800209e1b3ea4dcc806129b56, I found a two-call state bug in the documented name-collision replacement path.

servers_for_cli is pre-seeded from self.sdk_mcp_servers. If an existing SDK server named tools is replaced with an external config of the same name, the first wire payload is correct, but the old self.sdk_mcp_servers["tools"] entry (and bridge) remains registered.

A later authoritative set_mcp_servers({"other": ...}) then pre-seeds tools from that stale local registry and silently resurrects the old SDK server.

A regression can be two calls: replace SDK tools with an external tools, then call set_mcp_servers({"other": ...}) and assert tools is not reintroduced.

Could the local SDK registry / bridge state be updated to match caller replacements only after a successful mcp_set_servers response? Staging new local registrations until that response would also avoid Python/CLI divergence if the control request fails or times out.

[omgupta-duplo]

Good catch -- both confirmed and fixed in e438e0d.

The resurrection bug. You're right that only the wire payload reflected the replacement. A caller entry that is not an SDK config now also deregisters whatever it replaced:

else:
    # A non-SDK entry takes over this name for the session; any
    # in-process server registered under it is gone with it.
    removed.add(name)
    servers_for_cli[name] = config

so the next pre-seed can't re-emit it. One case I deliberately left alone: an SDK entry sent without an instance ({"type": "sdk", "name": "x"}) is a re-declaration of the server already registered under that name -- the same shape the pre-seed itself emits -- not a replacement, so it keeps its registration.

Staging until the response. Agreed, and done for the same reason you give. added/removed are collected during the loop and applied only after _send_control_request returns, so a rejected or timed-out request leaves the registry and bridges exactly as they were instead of diverging from the CLI.

Tests. Added the two-call regression you described -- replace SDK tools with an external tools, then set_mcp_servers({"other": ...}), and assert tools is not reintroduced in the second payload (and is gone from sdk_mcp_servers/_sdk_mcp_bridges). Also added a failure-path test asserting no partial local mutation when the request is rejected; _create_mock_transport_with_control_responses takes an optional fail_subtypes for that. I verified both fail against the previous commit and pass on this one.

Full file is green (76 passed), plus ruff and mypy.

[omgupta-duplo]

One more in 658bb0d, from re-reading the lifecycle around the fix above -- not something you raised, and half of it predates your review.

_close_impl stops in-process servers with for bridge in self._sdk_mcp_bridges.values(): await bridge.aclose(), so a bridge taken out of that dict is never stopped. _Session spawns its task with spawn_detached() rather than Query.spawn_task(), so it is not in _child_tasks either and the teardown task.cancel() loop does not reach it -- aclose() is the only stop signal. Two paths dropped a bridge without calling it:

• the removal I added above for a replaced name, and
• self._sdk_mcp_bridges[name] = SdkMcpBridge(name, instance) overwriting an existing name, which has been there since the first commit.

Once a bridge has served a message it owns a live session (a task running Server.run over a stream pair), so abandoning one leaves that task running for the rest of the process, never unwinds the server's lifespan, and never fails callers waiting on its pending responses. A bridge that never received a message is harmless -- _session is None and aclose() is a no-op.

Outgoing bridges are now collected while the new set is applied and closed after the swap, so traffic reaches the new bridge while the old one winds down -- the order SdkMcpBridge.handle already uses when it replaces a finished session. A failed wind-down is logged rather than raised, since the CLI has applied the set by then and the call should not fail.

Two tests cover it (replaced by an external server, and re-registered with a new instance); both fail on e438e0d and pass here. Full suite green: 1603 passed, 6 skipped, plus ruff and mypy.

Two things I noticed but deliberately left alone, happy to follow up if you'd rather they were part of this:

• There is no way to remove an SDK server. They are unconditionally pre-seeded, so the only way to get rid of one is to shadow its name with a non-SDK config.
• An SDK entry sent without an instance for a name that was never registered tells the CLI the server exists while _handle_sdk_mcp_request answers -32601 for every call to it. Rejecting that up front would be cheap.