diff --git a/src/fast_agent/agents/mcp_agent.py b/src/fast_agent/agents/mcp_agent.py
index 0fd86091..246304b2 100644
--- a/src/fast_agent/agents/mcp_agent.py
+++ b/src/fast_agent/agents/mcp_agent.py
@@ -668,13 +668,13 @@ async def run_tools(self, request: PromptMessageExtended) -> PromptMessageExtend
result = await self.call_tool(tool_name, tool_args)
tool_results[correlation_id] = result
- # Get skybridge config for this tool if available
+ # Show tool result (like ToolAgent does)
skybridge_config = None
if namespaced_tool:
- server_name = namespaced_tool.server_name
- skybridge_config = self._aggregator._skybridge_configs.get(server_name)
+ skybridge_config = await self._aggregator.get_skybridge_config(
+ namespaced_tool.server_name
+ )
- # Show tool result (like ToolAgent does)
self.display.show_tool_result(
name=self._name,
result=result,
diff --git a/src/fast_agent/mcp/mcp_aggregator.py b/src/fast_agent/mcp/mcp_aggregator.py
index 751acdb0..19511cf9 100644
--- a/src/fast_agent/mcp/mcp_aggregator.py
+++ b/src/fast_agent/mcp/mcp_aggregator.py
@@ -162,7 +162,7 @@ def __init__(
server_names: List[str],
connection_persistence: bool = True,
context: Optional["Context"] = None,
- name: str = None,
+ name: str | None = None,
config: Optional[Any] = None, # Accept the agent config for elicitation_handler access
**kwargs,
) -> None:
@@ -523,7 +523,8 @@ async def _evaluate_skybridge_for_server(
)
)
- supports_resources = await self.server_supports_feature(server_name, "resources")
+ raw_resources_capability = await self.server_supports_feature(server_name, "resources")
+ supports_resources = bool(raw_resources_capability)
config.supports_resources = supports_resources
config.tools = tool_configs
@@ -582,7 +583,10 @@ async def _evaluate_skybridge_for_server(
sky_resource.mime_type = seen_mime_types[0]
if not sky_resource.is_skybridge:
- warning = "ui:// detected but resource is not of type 'text/html+skybridge'"
+ observed_type = sky_resource.mime_type or "unknown MIME type"
+ warning = (
+ f"served as '{observed_type}' instead of '{SKYBRIDGE_MIME_TYPE}'"
+ )
sky_resource.warning = warning
config.warnings.append(f"{uri_str}: {warning}")
@@ -608,7 +612,8 @@ async def _evaluate_skybridge_for_server(
if not resource_match.is_skybridge:
warning = (
f"Tool '{tool_config.namespaced_tool_name}' references resource "
- f"'{resource_match.uri}' that is not Skybridge MIME type"
+ f"'{resource_match.uri}' served as '{resource_match.mime_type or 'unknown'}' "
+ f"instead of '{SKYBRIDGE_MIME_TYPE}'"
)
tool_config.warning = warning
config.warnings.append(warning)
@@ -691,7 +696,15 @@ async def server_supports_feature(self, server_name: str, feature: str) -> bool:
if not capabilities:
return False
- return getattr(capabilities, feature, False)
+ feature_value = getattr(capabilities, feature, False)
+ if isinstance(feature_value, bool):
+ return feature_value
+ if feature_value is None:
+ return False
+ try:
+ return bool(feature_value)
+ except Exception: # noqa: BLE001
+ return True
async def list_servers(self) -> List[str]:
"""Return the list of server names aggregated by this agent."""
@@ -995,6 +1008,12 @@ async def get_skybridge_configs(self) -> Dict[str, SkybridgeServerConfig]:
await self.load_servers()
return dict(self._skybridge_configs)
+ async def get_skybridge_config(self, server_name: str) -> SkybridgeServerConfig | None:
+ """Return the Skybridge configuration for a specific server, loading if necessary."""
+ if not self.initialized:
+ await self.load_servers()
+ return self._skybridge_configs.get(server_name)
+
async def _execute_on_server(
self,
server_name: str,
@@ -1002,7 +1021,7 @@ async def _execute_on_server(
operation_name: str,
method_name: str,
method_args: Dict[str, Any] = None,
- error_factory: Callable[[str], R] = None,
+ error_factory: Callable[[str], R] | None = None,
progress_callback: ProgressFnT | None = None,
) -> R:
"""
diff --git a/src/fast_agent/ui/console_display.py b/src/fast_agent/ui/console_display.py
index 0865bd56..eee7932f 100644
--- a/src/fast_agent/ui/console_display.py
+++ b/src/fast_agent/ui/console_display.py
@@ -1,6 +1,6 @@
from enum import Enum
from json import JSONDecodeError
-from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Set, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Set, Tuple, Union
from mcp.types import CallToolResult
from rich.panel import Panel
@@ -870,19 +870,18 @@ def _create_combined_separator_status(self, left_content: str, right_info: str =
console.console.print(combined, markup=self._markup)
console.console.print()
- def show_skybridge_summary(
- self,
- agent_name: str,
+ @staticmethod
+ def summarize_skybridge_configs(
configs: Mapping[str, "SkybridgeServerConfig"] | None,
- ) -> None:
- """Display Skybridge availability and warnings."""
- if configs is None:
- return
-
+ ) -> Tuple[List[Dict[str, Any]], List[str]]:
+ """Convert raw Skybridge configs into display-friendly summary data."""
server_rows: List[Dict[str, Any]] = []
warnings: List[str] = []
warning_seen: Set[str] = set()
+ if not configs:
+ return server_rows, warnings
+
def add_warning(message: str) -> None:
formatted = message.strip()
if not formatted:
@@ -931,6 +930,16 @@ def add_warning(message: str) -> None:
message = f"{server_name} {message}"
add_warning(message)
+ return server_rows, warnings
+
+ def show_skybridge_summary(
+ self,
+ agent_name: str,
+ configs: Mapping[str, "SkybridgeServerConfig"] | None,
+ ) -> None:
+ """Display Skybridge availability and warnings."""
+ server_rows, warnings = self.summarize_skybridge_configs(configs)
+
if not server_rows and not warnings:
return
@@ -943,7 +952,6 @@ def add_warning(message: str) -> None:
else:
for row in server_rows:
server_name = row["server_name"]
- config = row["config"]
resource_count = row["valid_resource_count"]
total_resource_count = row["total_resource_count"]
tool_infos = row["active_tools"]
@@ -985,14 +993,14 @@ def add_warning(message: str) -> None:
console.console.print(
(
"[dim] ▶ "
- f"{invalid_count} {invalid_word} detected with non-skybridge MIME type[/dim]"
+ f"[/dim][cyan]{invalid_count}[/cyan][dim] {invalid_word} detected with non-skybridge MIME type[/dim]"
),
markup=self._markup,
)
for warning_entry in warnings:
console.console.print(
- f"[yellow]skybridge warning[/yellow] {warning_entry}",
+ f"[dim red] ▶ [/dim red][red]warning[/red] [dim]{warning_entry}[/dim]",
markup=self._markup,
)
diff --git a/src/fast_agent/ui/mcp_display.py b/src/fast_agent/ui/mcp_display.py
index 129a137c..56391606 100644
--- a/src/fast_agent/ui/mcp_display.py
+++ b/src/fast_agent/ui/mcp_display.py
@@ -39,6 +39,7 @@ class Colours:
# Capability token states
TOKEN_ERROR = "bright_red"
TOKEN_WARNING = "bright_cyan"
+ TOKEN_CAUTION = "bright_yellow"
TOKEN_DISABLED = "dim"
TOKEN_HIGHLIGHTED = "bright_yellow"
TOKEN_ENABLED = "bright_green"
@@ -232,6 +233,18 @@ def _format_capability_shorthand(
else:
entries.append(("In", "blue", False))
+ skybridge_config = getattr(status, "skybridge", None)
+ if not skybridge_config:
+ entries.append(("Sk", False, False))
+ else:
+ has_warnings = bool(getattr(skybridge_config, "warnings", None))
+ if has_warnings:
+ entries.append(("Sk", "warn", False))
+ elif getattr(skybridge_config, "enabled", False):
+ entries.append(("Sk", True, False))
+ else:
+ entries.append(("Sk", False, False))
+
if status.roots_configured:
entries.append(("Ro", True, False))
else:
@@ -260,6 +273,8 @@ def token_style(supported, highlighted) -> str:
return Colours.TOKEN_ERROR
if supported == "blue":
return Colours.TOKEN_WARNING
+ if supported == "warn":
+ return Colours.TOKEN_CAUTION
if not supported:
return Colours.TOKEN_DISABLED
if highlighted:
@@ -652,6 +667,8 @@ def _render_channel_summary(status: ServerStatus, indent: str, total_width: int)
symbol = SYMBOL_ERROR
elif name == "ping":
symbol = SYMBOL_PING
+ elif is_stdio and name == "activity":
+ symbol = SYMBOL_STDIO_ACTIVITY
else:
symbol = SYMBOL_RESPONSE
footer.append(symbol, style=f"{color}")
diff --git a/tests/integration/skybridge/fastagent.config.yaml b/tests/integration/skybridge/fastagent.config.yaml
new file mode 100644
index 00000000..19572a79
--- /dev/null
+++ b/tests/integration/skybridge/fastagent.config.yaml
@@ -0,0 +1,22 @@
+default_model: passthrough
+
+logger:
+ level: "info"
+ progress_display: false
+ show_chat: false
+ show_tools: false
+
+mcp:
+ servers:
+ skybridge_valid:
+ command: "uv"
+ args: ["run", "skybridge_test_server.py", "valid"]
+ description: "Skybridge server with valid resource/tool pairing"
+ skybridge_invalid_mime:
+ command: "uv"
+ args: ["run", "skybridge_test_server.py", "invalid-mime"]
+ description: "Skybridge server exposing ui:// resource without Skybridge MIME type"
+ skybridge_missing_resource:
+ command: "uv"
+ args: ["run", "skybridge_test_server.py", "missing-resource"]
+ description: "Skybridge server with Skybridge resources but missing tool linkage"
diff --git a/tests/integration/skybridge/skybridge_test_server.py b/tests/integration/skybridge/skybridge_test_server.py
new file mode 100644
index 00000000..e9aea21a
--- /dev/null
+++ b/tests/integration/skybridge/skybridge_test_server.py
@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Skybridge-focused MCP test server exposing multiple scenarios."""
+
+from __future__ import annotations
+
+import argparse
+from typing import TYPE_CHECKING
+
+from mcp.server.fastmcp import FastMCP
+
+if TYPE_CHECKING:
+ from mcp.types import Tool as MCPTool
+
+SKYBRIDGE_MIME_TYPE = "text/html+skybridge"
+
+
+class SkybridgeTestServer(FastMCP):
+ """FastMCP server that decorates tool listings with Skybridge meta tags."""
+
+ def __init__(self, *args, tool_templates: dict[str, str] | None = None, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self._tool_templates = tool_templates or {}
+
+ async def list_tools(self) -> list[MCPTool]:
+ tools = await super().list_tools()
+ for tool in tools:
+ template = self._tool_templates.get(tool.name)
+ if template:
+ tool.meta = {"openai/outputTemplate": template}
+ return tools
+
+
+def build_valid_scenario() -> SkybridgeTestServer:
+ server = SkybridgeTestServer(
+ name="Skybridge Valid Scenario",
+ tool_templates={"render_valid_widget": "ui://skybridge/widget-valid"},
+ )
+
+ @server.tool(name="render_valid_widget", description="Return HTML for a valid Skybridge widget")
+ def render_valid_widget() -> str:
+ return "Valid Skybridge Widget
"
+
+ @server.resource(
+ "ui://skybridge/widget-valid",
+ description="Valid Skybridge UI resource",
+ mime_type=SKYBRIDGE_MIME_TYPE,
+ )
+ def valid_widget_resource() -> str:
+ return "Valid Skybridge Widget
"
+
+ return server
+
+
+def build_invalid_mime_scenario() -> SkybridgeTestServer:
+ server = SkybridgeTestServer(
+ name="Skybridge Invalid MIME Scenario",
+ tool_templates={"render_invalid_widget": "ui://skybridge/widget-invalid"},
+ )
+
+ @server.tool(
+ name="render_invalid_widget", description="Return HTML that lacks the Skybridge MIME type"
+ )
+ def render_invalid_widget() -> str:
+ return "Invalid MIME
"
+
+ @server.resource(
+ "ui://skybridge/widget-invalid",
+ description="Resource served with a non-Skybridge mime type",
+ mime_type="text/html",
+ )
+ def invalid_widget_resource() -> str:
+ return "Invalid MIME
"
+
+ return server
+
+
+def build_missing_resource_scenario() -> SkybridgeTestServer:
+ server = SkybridgeTestServer(
+ name="Skybridge Missing Resource Scenario",
+ tool_templates={"render_missing_widget": "ui://skybridge/widget-missing"},
+ )
+
+ @server.tool(
+ name="render_missing_widget",
+ description="Advertises a template that does not exist on the server",
+ )
+ def render_missing_widget() -> str:
+ return "Missing Resource
"
+
+ @server.resource(
+ "ui://skybridge/orphan-widget",
+ description="Orphaned Skybridge resource with no tool linkage",
+ mime_type=SKYBRIDGE_MIME_TYPE,
+ )
+ def orphan_widget_resource() -> str:
+ return "Orphan Widget
"
+
+ return server
+
+
+SCENARIO_BUILDERS = {
+ "valid": build_valid_scenario,
+ "invalid-mime": build_invalid_mime_scenario,
+ "missing-resource": build_missing_resource_scenario,
+}
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Skybridge MCP test server scenarios")
+ parser.add_argument(
+ "scenario",
+ choices=SCENARIO_BUILDERS.keys(),
+ help="Which Skybridge scenario to run",
+ )
+ args = parser.parse_args()
+
+ server_factory = SCENARIO_BUILDERS[args.scenario]
+ app = server_factory()
+ app.run(transport="stdio")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/integration/skybridge/test_skybridge_integration.py b/tests/integration/skybridge/test_skybridge_integration.py
new file mode 100644
index 00000000..c648e4a9
--- /dev/null
+++ b/tests/integration/skybridge/test_skybridge_integration.py
@@ -0,0 +1,117 @@
+import pytest
+
+from fast_agent.mcp.skybridge import SKYBRIDGE_MIME_TYPE
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_skybridge_valid_tool_and_resource(fast_agent):
+ fast = fast_agent
+
+ @fast.agent(
+ name="skybridge_valid_agent",
+ instruction="Exercise Skybridge detection for valid resources.",
+ model="passthrough",
+ servers=["skybridge_valid"],
+ )
+ async def agent_case():
+ async with fast.run() as app:
+ agent = app.skybridge_valid_agent
+ await agent.list_mcp_tools()
+ aggregator = agent._aggregator
+ configs = await aggregator.get_skybridge_configs()
+ config = configs["skybridge_valid"]
+
+ assert config.supports_resources is True
+ assert config.enabled is True
+ assert not config.warnings
+ assert len(config.ui_resources) == 1
+ resource = config.ui_resources[0]
+ assert resource.is_skybridge is True
+ assert resource.mime_type == SKYBRIDGE_MIME_TYPE
+
+ assert len(config.tools) == 1
+ tool = config.tools[0]
+ assert tool.is_valid is True
+ assert tool.resource_uri == resource.uri
+
+ await agent_case()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_skybridge_invalid_mime_generates_warning(fast_agent):
+ fast = fast_agent
+
+ @fast.agent(
+ name="skybridge_invalid_mime_agent",
+ instruction="Skybridge detection with invalid MIME type.",
+ model="passthrough",
+ servers=["skybridge_invalid_mime"],
+ )
+ async def agent_case():
+ async with fast.run() as app:
+ agent = app.skybridge_invalid_mime_agent
+ await agent.list_mcp_tools()
+ aggregator = agent._aggregator
+ configs = await aggregator.get_skybridge_configs()
+ config = configs["skybridge_invalid_mime"]
+
+ assert config.supports_resources is True
+ assert config.enabled is False
+ assert config.ui_resources, "Expected to discover the ui:// resource"
+ resource = config.ui_resources[0]
+ assert resource.is_skybridge is False
+ assert (
+ resource.warning
+ == "served as 'text/html' instead of 'text/html+skybridge'"
+ )
+
+ assert config.tools, "Expected to capture the tool metadata"
+ tool = config.tools[0]
+ assert tool.is_valid is False
+ assert tool.warning is not None
+ assert "served as 'text/html' instead of 'text/html+skybridge'" in tool.warning
+ assert any(
+ "served as 'text/html' instead of 'text/html+skybridge'" in warning
+ for warning in config.warnings
+ )
+
+ await agent_case()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_skybridge_missing_resource_warns_and_flags_tools(fast_agent):
+ fast = fast_agent
+
+ @fast.agent(
+ name="skybridge_missing_resource_agent",
+ instruction="Skybridge detection with missing resource linkage.",
+ model="passthrough",
+ servers=["skybridge_missing_resource"],
+ )
+ async def agent_case():
+ async with fast.run() as app:
+ agent = app.skybridge_missing_resource_agent
+ await agent.list_mcp_tools()
+ aggregator = agent._aggregator
+ configs = await aggregator.get_skybridge_configs()
+ config = configs["skybridge_missing_resource"]
+
+ assert config.enabled is True, "Valid resource should mark server as enabled"
+ assert config.ui_resources, "Expected at least one Skybridge resource"
+ assert any(
+ "references missing Skybridge resource" in warning for warning in config.warnings
+ )
+ assert any(
+ "no tools expose them" in warning.lower() for warning in config.warnings
+ )
+
+ assert config.tools, "Expected to capture tool metadata"
+ tool = config.tools[0]
+ assert tool.is_valid is False
+ assert tool.warning is not None
+ assert "references missing Skybridge resource" in tool.warning
+
+ await agent_case()
diff --git a/tests/unit/fast_agent/mcp/test_mcp_aggregator_skybridge.py b/tests/unit/fast_agent/mcp/test_mcp_aggregator_skybridge.py
index a9522b89..582a2669 100644
--- a/tests/unit/fast_agent/mcp/test_mcp_aggregator_skybridge.py
+++ b/tests/unit/fast_agent/mcp/test_mcp_aggregator_skybridge.py
@@ -127,14 +127,20 @@ def test_skybridge_detection_warns_on_invalid_mime() -> None:
assert len(config.ui_resources) == 1
assert (
config.ui_resources[0].warning
- == "ui:// detected but resource is not of type 'text/html+skybridge'"
+ == "served as 'text/html' instead of 'text/html+skybridge'"
)
assert config.warnings
- assert "ui://component/app" in config.warnings[0]
+ assert config.warnings[0] == (
+ "ui://component/app: served as 'text/html' instead of 'text/html+skybridge'"
+ )
assert len(config.tools) == 1
tool_cfg = config.tools[0]
assert tool_cfg.is_valid is False
- assert tool_cfg.warning is not None
+ assert (
+ tool_cfg.warning
+ == "Tool 'test.tool_a' references resource 'ui://component/app' served as 'text/html' "
+ "instead of 'text/html+skybridge'"
+ )
aggregator._list_resources_from_server.assert_awaited_once_with( # type: ignore[attr-defined]
"test", check_support=False
)
diff --git a/tests/unit/fast_agent/ui/test_console_display_skybridge.py b/tests/unit/fast_agent/ui/test_console_display_skybridge.py
new file mode 100644
index 00000000..14fc48c7
--- /dev/null
+++ b/tests/unit/fast_agent/ui/test_console_display_skybridge.py
@@ -0,0 +1,57 @@
+from fast_agent.mcp.skybridge import (
+ SkybridgeResourceConfig,
+ SkybridgeServerConfig,
+ SkybridgeToolConfig,
+)
+from fast_agent.ui.console_display import ConsoleDisplay
+
+
+def test_summarize_skybridge_configs_flags_invalid_resource() -> None:
+ resource_warning = "served as 'text/html' instead of 'text/html+skybridge'"
+ resource = SkybridgeResourceConfig(
+ uri="ui://widget/pizza-map.html",
+ mime_type="text/html",
+ is_skybridge=False,
+ warning=resource_warning,
+ )
+ tool_warning = (
+ "Tool 'hf/pizzaz-pizza-map' references resource 'ui://widget/pizza-map.html' "
+ "served as 'text/html' instead of 'text/html+skybridge'"
+ )
+ tool = SkybridgeToolConfig(
+ tool_name="pizzaz-pizza-map",
+ namespaced_tool_name="hf/pizzaz-pizza-map",
+ template_uri="ui://widget/pizza-map.html",
+ is_valid=False,
+ warning=tool_warning,
+ )
+ config = SkybridgeServerConfig(
+ server_name="hf",
+ supports_resources=True,
+ ui_resources=[resource],
+ warnings=[f"{resource.uri}: {resource_warning}", tool_warning],
+ tools=[tool],
+ )
+
+ rows, warnings = ConsoleDisplay.summarize_skybridge_configs({"hf": config})
+
+ assert len(rows) == 1
+ row = rows[0]
+ assert row["server_name"] == "hf"
+ assert row["enabled"] is False
+ assert row["valid_resource_count"] == 0
+ assert row["total_resource_count"] == 1
+ assert row["active_tools"] == []
+
+ assert len(warnings) == 2
+ assert any("ui://widget/pizza-map.html" in warning for warning in warnings)
+ assert any("pizzaz-pizza-map" in warning for warning in warnings)
+
+
+def test_summarize_skybridge_configs_ignores_servers_without_signals() -> None:
+ config = SkybridgeServerConfig(server_name="empty")
+
+ rows, warnings = ConsoleDisplay.summarize_skybridge_configs({"empty": config})
+
+ assert rows == []
+ assert warnings == []