feat(mcp): expose MCP server status across all protocols

src/agentpool_server/acp_server/acp_agent.py

@@ -412,7 +412,19 @@ async def initialize(self, params: InitializeRequest) -> InitializeResponse:
         # Gate turn_complete advertisement on client's declared support
         client_caps = params.client_capabilities
         turn_complete = bool(client_caps.turn_complete) if client_caps is not None else False
-        return InitializeResponse.create(
+        # Collect MCP server statuses for client visibility
+        mcp_servers_meta: list[dict[str, Any]] = []
+        if pool is not None and pool.mcp is not None:
+            for provider in pool.mcp.get_mcp_providers():
+                status = provider.get_status()
+                mcp_servers_meta.append({
+                    "name": status.name,
+                    "status": status.status,
+                    "display_name": status.display_name,
+                    "server_type": status.server_type,
+                    "error": status.error,
+                })
+        response = InitializeResponse.create(
             protocol_version=version,
             name="agentpool",
             title="AgentPool",
@@ -431,6 +443,9 @@ async def initialize(self, params: InitializeRequest) -> InitializeResponse:
             providers=True,
             turn_complete=turn_complete,
         )
+        if mcp_servers_meta:
+            response.field_meta = {"mcp_servers": mcp_servers_meta}
+        return response
 
     async def new_session(self, params: NewSessionRequest) -> NewSessionResponse:
         """Create a new session."""

src/agentpool_server/mcp_server/server.py

@@ -167,6 +167,35 @@ async def handler(**kwargs: Any) -> str:
         self._prompts_registered = True
         logger.info("Registered MCP prompts", count=len(prompts))
 
+    def _register_mcp_status_resource(self) -> None:
+        """Register a resource exposing MCP server connection statuses.
+
+        Clients can read ``agentpool://mcp-servers/status`` to get a JSON
+        summary of all configured MCP servers and their connection state.
+        """
+        import json
+
+        @self.fastmcp.resource(
+            uri="agentpool://mcp-servers/status",
+            name="MCPServersStatus",
+            description="Connection status of all MCP servers managed by AgentPool.",
+            mime_type="application/json",
+        )
+        async def get_mcp_servers_status() -> str:
+            """Return JSON with each MCP server's connection status."""
+            mcp_manager = self.pool.mcp
+            statuses: list[dict[str, Any]] = []
+            for provider in mcp_manager.get_mcp_providers():
+                status = provider.get_status()
+                statuses.append({
+                    "name": status.name,
+                    "status": status.status,
+                    "display_name": status.display_name,
+                    "server_type": status.server_type,
+                    "error": status.error,
+                })
+            return json.dumps({"servers": statuses})
+
     def _register_resource_handlers(self) -> None:
         """Register resource subscription handlers."""
 
@@ -226,6 +255,7 @@ async def _start_async(self) -> None:
         # Register pool tools and prompts before starting
         await self._register_pool_tools()
         await self._register_pool_prompts()
+        self._register_mcp_status_resource()
 
         # Start FastMCP server
         if self.config.transport == "stdio":

src/agentpool_server/opencode_server/converters.py

@@ -87,6 +87,7 @@ def to_mcp_status(status: MCPServerStatus) -> MCPStatus:
         display_name=status.display_name or status.name,
         status=to_opencode_mcp_status(status.status),
         error=status.error,
+        server_type=status.server_type,
     )
 
 

src/agentpool_server/opencode_server/models/mcp.py

@@ -33,6 +33,9 @@ class MCPStatus(OpenCodeBaseModel):
     tools: list[str] = Field(default_factory=list)
     error: str | None = None
 
+    server_type: str = "unknown"
+    """Transport type (stdio, sse, http, acp)."""
+
 
 class McpAuthorizationResponse(OpenCodeBaseModel):
     """Response from starting MCP OAuth flow."""

src/agentpool_server/opencode_server/routes/config_routes.py

@@ -15,6 +15,7 @@
 from agentpool_server.opencode_server.dependencies import StateDep
 from agentpool_server.opencode_server.models import (
     Config,
+    MCPStatus,
     Mode,
     Model,
     ModelCost,
@@ -516,3 +517,21 @@ async def list_modes(state: StateDep) -> list[Mode]:
         )
         for mode in category.available_modes
     ]
+
+
+@router.get("/config/mcp-servers")
+async def get_mcp_servers(state: StateDep) -> dict[str, MCPStatus]:
+    """Get MCP server connection statuses.
+
+    Returns a dict mapping server names to their connection status,
+    display name, transport type, and any error messages. This endpoint
+    complements ``GET /mcp`` by following the ``/config/*`` route
+    convention for configuration-adjacent queries.
+    """
+    from agentpool_server.opencode_server.converters import to_mcp_status
+
+    if state.agent is None:
+        return {}
+
+    server_info = await state.agent.get_mcp_server_info()
+    return {name: to_mcp_status(status) for name, status in server_info.items()}

tests/agents/native_agent/test_get_agentlet_capabilities.py

@@ -639,6 +639,7 @@ async def test_from_config_capabilities_not_duplicated() -> None:
     This test calls from_config() with a config containing a capability, then
     calls get_agentlet() and verifies the capability appears exactly once.
     """
+    from llmling_models_config import TestModelConfig
     from pydantic_ai.capabilities import Instrumentation
 
     from agentpool.models.agents import NativeAgentConfig
@@ -650,7 +651,7 @@ async def test_from_config_capabilities_not_duplicated() -> None:
     )
     config = NativeAgentConfig(
         name="test_dedup_agent",
-        model="openai:gpt-4o-mini",
+        model=TestModelConfig(custom_output_text="test"),
         system_prompt=["Be helpful."],
         capabilities=[cap_config],
     )

tests/servers/acp_server/test_acp_initialize_mcp_meta.py

@@ -0,0 +1,88 @@
+"""Tests for ACP initialize response MCP server status metadata.
+
+Verifies that AgentPoolACPAgent.initialize() includes MCP server connection
+statuses in the response's field_meta.mcp_servers.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import Mock
+
+import pytest
+
+from acp import InitializeRequest
+from agentpool.common_types import MCPServerStatus
+
+
+if TYPE_CHECKING:
+    from agentpool_server.acp_server.acp_agent import AgentPoolACPAgent
+
+
+pytestmark = pytest.mark.asyncio
+
+
+def _make_mock_provider(status: MCPServerStatus) -> Mock:
+    """Create a mock MCPResourceProvider with the given status."""
+    provider = Mock()
+    provider.get_status = Mock(return_value=status)
+    return provider
+
+
+async def test_initialize_includes_mcp_servers_in_field_meta(
+    mock_acp_agent: AgentPoolACPAgent,
+) -> None:
+    """initialize() response includes mcp_servers in field_meta."""
+    pool = mock_acp_agent.agent_pool
+    assert pool is not None
+
+    providers = [
+        _make_mock_provider(
+            MCPServerStatus(
+                name="srv-1",
+                status="connected",
+                server_type="stdio",
+                display_name="Server One",
+            ),
+        ),
+        _make_mock_provider(
+            MCPServerStatus(
+                name="srv-2",
+                status="error",
+                server_type="sse",
+                display_name="Server Two",
+                error="Connection refused",
+            ),
+        ),
+    ]
+    pool.mcp.get_mcp_providers = Mock(return_value=providers)
+
+    response = await mock_acp_agent.initialize(
+        InitializeRequest(protocol_version=1),
+    )
+
+    assert response.field_meta is not None
+    assert "mcp_servers" in response.field_meta
+    servers = response.field_meta["mcp_servers"]
+    assert len(servers) == 2
+    assert servers[0]["name"] == "srv-1"
+    assert servers[0]["status"] == "connected"
+    assert servers[0]["server_type"] == "stdio"
+    assert servers[1]["name"] == "srv-2"
+    assert servers[1]["status"] == "error"
+    assert servers[1]["error"] == "Connection refused"
+
+
+async def test_initialize_field_meta_none_when_no_providers(
+    mock_acp_agent: AgentPoolACPAgent,
+) -> None:
+    """initialize() response has field_meta None when no MCP servers configured."""
+    pool = mock_acp_agent.agent_pool
+    assert pool is not None
+    pool.mcp.get_mcp_providers = Mock(return_value=[])
+
+    response = await mock_acp_agent.initialize(
+        InitializeRequest(protocol_version=1),
+    )
+
+    assert response.field_meta is None

tests/servers/mcp_server/__init__.py

@@ -0,0 +1 @@
+"""Tests for MCP server protocol."""

tests/servers/mcp_server/test_mcp_status_resource.py

@@ -0,0 +1,112 @@
+"""Tests for the agentpool://mcp-servers/status MCP resource.
+
+Tests that the MCPServer registers a resource exposing MCP server
+connection statuses as JSON.
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import Mock
+
+import pytest
+
+from agentpool.common_types import MCPServerStatus
+
+
+pytestmark = pytest.mark.asyncio
+
+
+def _make_mock_pool_with_providers(
+    providers: list[MCPResourceProviderStub],
+) -> Mock:
+    """Create a mock AgentPool with the given MCP providers."""
+    pool = Mock()
+    mcp_manager = Mock()
+    mcp_manager.get_mcp_providers = Mock(return_value=providers)
+    pool.mcp = mcp_manager
+    return pool
+
+
+class MCPResourceProviderStub:
+    """Minimal stub matching the interface used by the resource handler."""
+
+    def __init__(self, status: MCPServerStatus) -> None:
+        self._status = status
+
+    def get_status(self) -> MCPServerStatus:
+        return self._status
+
+
+async def test_mcp_status_resource_returns_json() -> None:
+    """The status resource returns JSON with server connection info."""
+    from agentpool_config.pool_server import MCPPoolServerConfig
+    from agentpool_server.mcp_server.server import MCPServer
+
+    providers = [
+        MCPResourceProviderStub(
+            MCPServerStatus(
+                name="srv-1",
+                status="connected",
+                server_type="stdio",
+                display_name="Server One",
+            ),
+        ),
+        MCPResourceProviderStub(
+            MCPServerStatus(
+                name="srv-2",
+                status="error",
+                server_type="sse",
+                display_name="Server Two",
+                error="Connection refused",
+            ),
+        ),
+    ]
+    pool = _make_mock_pool_with_providers(providers)
+    config = MCPPoolServerConfig(enabled=True, transport="stdio")
+    server = MCPServer(pool, config)
+
+    server._register_mcp_status_resource()
+
+    fastmcp = server.fastmcp
+    resources = await fastmcp.list_resources()
+    status_resource = next(
+        (r for r in resources if str(r.uri) == "agentpool://mcp-servers/status"),
+        None,
+    )
+    assert status_resource is not None
+    assert status_resource.mime_type == "application/json"
+    read_result = await fastmcp.read_resource(
+        "agentpool://mcp-servers/status",
+        run_middleware=False,
+    )
+    content = read_result.contents[0].content
+    data = json.loads(content)
+    assert "servers" in data
+    servers = data["servers"]
+    assert len(servers) == 2
+    assert servers[0]["name"] == "srv-1"
+    assert servers[0]["status"] == "connected"
+    assert servers[0]["server_type"] == "stdio"
+    assert servers[1]["name"] == "srv-2"
+    assert servers[1]["status"] == "error"
+    assert servers[1]["error"] == "Connection refused"
+
+
+async def test_mcp_status_resource_empty_when_no_providers() -> None:
+    """The status resource returns empty servers list when no providers exist."""
+    from agentpool_config.pool_server import MCPPoolServerConfig
+    from agentpool_server.mcp_server.server import MCPServer
+
+    pool = _make_mock_pool_with_providers([])
+    config = MCPPoolServerConfig(enabled=True, transport="stdio")
+    server = MCPServer(pool, config)
+
+    server._register_mcp_status_resource()
+
+    read_result = await server.fastmcp.read_resource(
+        "agentpool://mcp-servers/status",
+        run_middleware=False,
+    )
+    data = json.loads(read_result.contents[0].content)
+    assert data["servers"] == []