feat: add typed McpServerStatus return for get_mcp_status

Author: qing-ant

src/claude_agent_sdk/__init__.py

@@ -31,6 +31,13 @@
     HookMatcher,
     McpSdkServerConfig,
     McpServerConfig,
+    McpServerConnectionStatus,
+    McpServerInfo,
+    McpServerStatus,
+    McpServerStatusConfig,
+    McpStatusResponse,
+    McpToolAnnotations,
+    McpToolInfo,
     Message,
     NotificationHookInput,
     NotificationHookSpecificOutput,
@@ -330,6 +337,13 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
     "PermissionMode",
     "McpServerConfig",
     "McpSdkServerConfig",
+    "McpServerStatus",
+    "McpServerStatusConfig",
+    "McpServerConnectionStatus",
+    "McpServerInfo",
+    "McpStatusResponse",
+    "McpToolAnnotations",
+    "McpToolInfo",
     "UserMessage",
     "AssistantMessage",
     "SystemMessage",

src/claude_agent_sdk/client.py

@@ -8,7 +8,14 @@
 
 from . import Transport
 from ._errors import CLIConnectionError
-from .types import ClaudeAgentOptions, HookEvent, HookMatcher, Message, ResultMessage
+from .types import (
+    ClaudeAgentOptions,
+    HookEvent,
+    HookMatcher,
+    McpStatusResponse,
+    Message,
+    ResultMessage,
+)
 
 
 class ClaudeSDKClient:
@@ -375,30 +382,37 @@ async def stop_task(self, task_id: str) -> None:
             raise CLIConnectionError("Not connected. Call connect() first.")
         await self._query.stop_task(task_id)
 
-    async def get_mcp_status(self) -> dict[str, Any]:
+    async def get_mcp_status(self) -> McpStatusResponse:
         """Get current MCP server connection status (only works with streaming mode).
 
         Queries the Claude Code CLI for the live connection status of all
         configured MCP servers.
 
         Returns:
-            Dictionary with MCP server status information. Contains a
-            'mcpServers' key with a list of server status objects, each having:
+            McpStatusResponse dictionary with an 'mcpServers' key containing
+            a list of McpServerStatus entries. Each entry includes:
             - 'name': Server name (str)
             - 'status': Connection status ('connected', 'pending', 'failed',
               'needs-auth', 'disabled')
+            - 'serverInfo': MCP server name/version (when connected)
+            - 'error': Error message (when status is 'failed')
+            - 'config': Server configuration (stdio/sse/http/sdk/claudeai-proxy)
+            - 'scope': Configuration scope (e.g., project, user, local)
+            - 'tools': List of tools provided by the server (when connected)
 
         Example:
             ```python
             async with ClaudeSDKClient(options) as client:
                 status = await client.get_mcp_status()
-                for server in status.get("mcpServers", []):
+                for server in status["mcpServers"]:
                     print(f"{server['name']}: {server['status']}")
+                    if server["status"] == "failed":
+                        print(f"  Error: {server.get('error')}")
             ```
         """
         if not self._query:
             raise CLIConnectionError("Not connected. Call connect() first.")
-        result: dict[str, Any] = await self._query.get_mcp_status()
+        result: McpStatusResponse = await self._query.get_mcp_status()
         return result
 
     async def get_server_info(self) -> dict[str, Any] | None:

src/claude_agent_sdk/types.py

@@ -504,6 +504,116 @@ class McpSdkServerConfig(TypedDict):
 )
 
 
+# MCP Server Status types (returned by get_mcp_status)
+# These mirror the TypeScript SDK's McpServerStatus type and use wire-format
+# field names (camelCase where applicable) since they come directly from CLI
+# JSON output.
+
+
+class McpSdkServerConfigStatus(TypedDict):
+    """SDK MCP server config as returned in status responses.
+
+    Unlike McpSdkServerConfig (which includes the in-process `instance`),
+    this output-only type only has serializable fields.
+    """
+
+    type: Literal["sdk"]
+    name: str
+
+
+class McpClaudeAIProxyServerConfig(TypedDict):
+    """Claude.ai proxy MCP server config.
+
+    Output-only type that appears in status responses for servers proxied
+    through Claude.ai.
+    """
+
+    type: Literal["claudeai-proxy"]
+    url: str
+    id: str
+
+
+# Broader config type for status responses (includes claudeai-proxy which is
+# output-only)
+McpServerStatusConfig = (
+    McpStdioServerConfig
+    | McpSSEServerConfig
+    | McpHttpServerConfig
+    | McpSdkServerConfigStatus
+    | McpClaudeAIProxyServerConfig
+)
+
+
+class McpToolAnnotations(TypedDict, total=False):
+    """Tool annotations as returned in MCP server status.
+
+    Wire format uses camelCase field names (from CLI JSON output).
+    """
+
+    readOnly: bool
+    destructive: bool
+    openWorld: bool
+
+
+class McpToolInfo(TypedDict):
+    """Information about a tool provided by an MCP server."""
+
+    name: str
+    description: NotRequired[str]
+    annotations: NotRequired[McpToolAnnotations]
+
+
+class McpServerInfo(TypedDict):
+    """Server info from MCP initialize handshake (available when connected)."""
+
+    name: str
+    version: str
+
+
+# Connection status values for an MCP server
+McpServerConnectionStatus = Literal[
+    "connected", "failed", "needs-auth", "pending", "disabled"
+]
+
+
+class McpServerStatus(TypedDict):
+    """Status information for an MCP server connection.
+
+    Returned by `ClaudeSDKClient.get_mcp_status()` in the `mcpServers` list.
+    """
+
+    name: str
+    """Server name as configured."""
+
+    status: McpServerConnectionStatus
+    """Current connection status."""
+
+    serverInfo: NotRequired[McpServerInfo]
+    """Server information from MCP handshake (available when connected)."""
+
+    error: NotRequired[str]
+    """Error message (available when status is 'failed')."""
+
+    config: NotRequired[McpServerStatusConfig]
+    """Server configuration (includes URL for HTTP/SSE servers)."""
+
+    scope: NotRequired[str]
+    """Configuration scope (e.g., project, user, local, claudeai, managed)."""
+
+    tools: NotRequired[list[McpToolInfo]]
+    """Tools provided by this server (available when connected)."""
+
+
+class McpStatusResponse(TypedDict):
+    """Response from `ClaudeSDKClient.get_mcp_status()`.
+
+    Wraps the list of server statuses under the `mcpServers` key, matching
+    the wire-format response shape.
+    """
+
+    mcpServers: list[McpServerStatus]
+
+
 class SdkPluginConfig(TypedDict):
     """SDK plugin configuration.
 

tests/test_streaming_client.py

@@ -682,6 +682,152 @@ async def _test():
 
         anyio.run(_test)
 
+    def test_get_mcp_status(self):
+        """Test get_mcp_status returns McpStatusResponse shape."""
+
+        async def _test():
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = AsyncMock()
+                mock_transport.connect = AsyncMock()
+                mock_transport.close = AsyncMock()
+                mock_transport.end_input = AsyncMock()
+                mock_transport.is_ready = Mock(return_value=True)
+                mock_transport_class.return_value = mock_transport
+
+                written_messages: list[str] = []
+
+                async def mock_write(data):
+                    written_messages.append(data)
+
+                mock_transport.write = AsyncMock(side_effect=mock_write)
+
+                # Simulated mcp_status response matching McpServerStatus shape
+                mcp_status_response = {
+                    "mcpServers": [
+                        {
+                            "name": "my-http-server",
+                            "status": "connected",
+                            "serverInfo": {
+                                "name": "my-http-server",
+                                "version": "1.0.0",
+                            },
+                            "config": {
+                                "type": "http",
+                                "url": "https://example.com/mcp",
+                            },
+                            "scope": "project",
+                            "tools": [
+                                {
+                                    "name": "greet",
+                                    "description": "Greet a user",
+                                    "annotations": {"readOnly": True},
+                                },
+                                {"name": "reset"},
+                            ],
+                        },
+                        {
+                            "name": "failed-server",
+                            "status": "failed",
+                            "error": "Connection refused",
+                        },
+                        {
+                            "name": "proxy-server",
+                            "status": "needs-auth",
+                            "config": {
+                                "type": "claudeai-proxy",
+                                "url": "https://claude.ai/proxy",
+                                "id": "proxy-123",
+                            },
+                        },
+                    ]
+                }
+
+                async def control_protocol_generator():
+                    last_check = 0
+                    timeout_counter = 0
+                    while timeout_counter < 200:
+                        await asyncio.sleep(0.01)
+                        timeout_counter += 1
+
+                        for msg_str in written_messages[last_check:]:
+                            try:
+                                msg = json.loads(msg_str.strip())
+                                if msg.get("type") == "control_request":
+                                    subtype = msg.get("request", {}).get("subtype")
+                                    if subtype == "initialize":
+                                        yield {
+                                            "type": "control_response",
+                                            "response": {
+                                                "request_id": msg.get("request_id"),
+                                                "subtype": "success",
+                                                "response": {},
+                                            },
+                                        }
+                                    elif subtype == "mcp_status":
+                                        yield {
+                                            "type": "control_response",
+                                            "response": {
+                                                "request_id": msg.get("request_id"),
+                                                "subtype": "success",
+                                                "response": mcp_status_response,
+                                            },
+                                        }
+                            except (json.JSONDecodeError, KeyError, AttributeError):
+                                pass
+                        last_check = len(written_messages)
+
+                mock_transport.read_messages = control_protocol_generator
+
+                async with ClaudeSDKClient() as client:
+                    status = await client.get_mcp_status()
+
+                    # Verify response conforms to McpStatusResponse shape
+                    assert "mcpServers" in status
+                    servers = status["mcpServers"]
+                    assert len(servers) == 3
+
+                    # Connected server with full info
+                    connected = servers[0]
+                    assert connected["name"] == "my-http-server"
+                    assert connected["status"] == "connected"
+                    assert connected["serverInfo"]["version"] == "1.0.0"
+                    assert connected["config"]["type"] == "http"
+                    assert connected["config"]["url"] == "https://example.com/mcp"
+                    assert connected["scope"] == "project"
+                    assert len(connected["tools"]) == 2
+                    assert connected["tools"][0]["name"] == "greet"
+                    assert connected["tools"][0]["annotations"]["readOnly"] is True
+                    # Tool without optional fields
+                    assert connected["tools"][1]["name"] == "reset"
+                    assert "description" not in connected["tools"][1]
+
+                    # Failed server with error
+                    failed = servers[1]
+                    assert failed["name"] == "failed-server"
+                    assert failed["status"] == "failed"
+                    assert failed["error"] == "Connection refused"
+
+                    # Server with claudeai-proxy config
+                    proxy = servers[2]
+                    assert proxy["name"] == "proxy-server"
+                    assert proxy["status"] == "needs-auth"
+                    assert proxy["config"]["type"] == "claudeai-proxy"
+                    assert proxy["config"]["id"] == "proxy-123"
+
+        anyio.run(_test)
+
+    def test_get_mcp_status_not_connected(self):
+        """Test get_mcp_status when not connected raises error."""
+
+        async def _test():
+            client = ClaudeSDKClient()
+            with pytest.raises(CLIConnectionError, match="Not connected"):
+                await client.get_mcp_status()
+
+        anyio.run(_test)
+
     def test_client_with_options(self):
         """Test client initialization with options."""