diff --git a/README.md b/README.md index 21f6a5448..e55329a97 100644 --- a/README.md +++ b/README.md @@ -592,6 +592,36 @@ lost_baggage = SwarmAgent( +### Icons + +You can add icons to your agents for better visualization in UIs that support it (e.g. Claude Desktop). +To do so, you can add an `icons` parameter to your `MCPApp` or `@app.tool` definitions. + +```python +# load an icon from file and serve it as a data URI +icon_path = Path("my-icon.png") +icon_data = base64.standard_b64encode(icon_path.read_bytes()).decode() +icon_data_uri = f"data:image/png;base64,{icon_data}" +icon = Icon(src=icon_data_uri, mimeType="image/png", sizes=["64x64"]) + +# alternatively, one can use an external URL: +web_icon = Icon(src="https://example.com/my-icon.png", mimeType="image/png", sizes=["64x64"]) + +app = MCPApp( + name="my_app_with_icon", + description="Agent server with an icon", + icons=[icon], + # ... +) + + +@app.tool(icons=[icon]) +async def my_tool(...) -> ...: + # ... +``` + +If you inject your own `FastMCP` instance into an `MCPApp`, you will have to add your icons there. + ### App Config Create an [`mcp_agent.config.yaml`](/schema/mcp-agent.config.schema.json) and define secrets via either a gitignored [`mcp_agent.secrets.yaml`](./examples/basic/mcp_basic_agent/mcp_agent.secrets.yaml.example) or a local [`.env`](./examples/basic/mcp_basic_agent/.env.example). In production, prefer `MCP_APP_SETTINGS_PRELOAD` to avoid writing plaintext secrets to disk. diff --git a/examples/mcp_agent_server/temporal/mag.png b/examples/mcp_agent_server/temporal/mag.png new file mode 100644 index 000000000..7d08c1986 Binary files /dev/null and b/examples/mcp_agent_server/temporal/mag.png differ diff --git a/examples/mcp_agent_server/temporal/main.py b/examples/mcp_agent_server/temporal/main.py index 659eb1af4..851c3ff86 100644 --- a/examples/mcp_agent_server/temporal/main.py +++ b/examples/mcp_agent_server/temporal/main.py @@ -9,8 +9,10 @@ """ import asyncio +import base64 import logging import os +from pathlib import Path from mcp.types import Icon, ModelHint, ModelPreferences, SamplingMessage, TextContent from temporalio.exceptions import ApplicationError @@ -93,12 +95,17 @@ async def run( return WorkflowResult(value=result) +icon_file = Path(__file__).parent / "mag.png" +icon_data = base64.standard_b64encode(icon_file.read_bytes()).decode() +icon_data_uri = f"data:image/png;base64,{icon_data}" +mag_icon = Icon(src=icon_data_uri, mimeType="image/png", sizes=["64x64"]) + @app.tool( name="finder_tool", title="Finder Tool", description="Run the Finder workflow synchronously.", annotations={"idempotentHint": False}, - icons=[Icon(src="emoji:mag")], + icons=[mag_icon], meta={"category": "demo", "engine": "temporal"}, structured_output=False, ) diff --git a/src/mcp_agent/app.py b/src/mcp_agent/app.py index 71fee25c7..e7c04afd3 100644 --- a/src/mcp_agent/app.py +++ b/src/mcp_agent/app.py @@ -23,7 +23,6 @@ from mcp import ServerSession from mcp.server.fastmcp import FastMCP from mcp.types import ToolAnnotations, Icon - from mcp_agent.core.context import Context, initialize_context, cleanup_context from mcp_agent.config import Settings, get_settings from mcp_agent.executor.signal_registry import SignalRegistry @@ -56,6 +55,11 @@ P = ParamSpec("P") R = TypeVar("R") +phetch = Icon( + src="https://s3.us-east-1.amazonaws.com/publicdata.lastmileai.com/phetch.png", + mimeType="image/png", + sizes=["48x48"], +) class MCPApp: """ @@ -89,6 +93,7 @@ def __init__( signal_notification: SignalWaitCallback | None = None, upstream_session: Optional["ServerSession"] = None, model_selector: ModelSelector | None = None, + icons: list[Icon] | None = None, session_id: str | None = None, ): """ @@ -140,6 +145,10 @@ def __init__( self._signal_notification = signal_notification self._upstream_session = upstream_session self._model_selector = model_selector + if icons: + self._icons = icons + else: + self._icons = [phetch] self._session_id_override = session_id self._workflows: Dict[str, Type["Workflow"]] = {} # id to workflow class @@ -954,6 +963,8 @@ def decorator(fn: Callable[P, R]) -> Callable[P, R]: icons_list.append(Icon(**icon)) else: raise TypeError("icons entries must be Icon or mapping") + else: + icons_list = [phetch] meta_payload: Dict[str, Any] | None = None if meta is not None: @@ -1062,6 +1073,8 @@ def decorator(fn: Callable[P, R]) -> Callable[P, R]: icons_list.append(Icon(**icon)) else: raise TypeError("icons entries must be Icon or mapping") + else: + icons_list = [phetch] meta_payload: Dict[str, Any] | None = None if meta is not None: @@ -1073,6 +1086,7 @@ def decorator(fn: Callable[P, R]) -> Callable[P, R]: description=description, mark_sync_tool=False, ) + # Defer alias tool registration for run/get_status self._declared_tools.append( { diff --git a/src/mcp_agent/data/phetch.png b/src/mcp_agent/data/phetch.png new file mode 100644 index 000000000..229681986 Binary files /dev/null and b/src/mcp_agent/data/phetch.png differ diff --git a/src/mcp_agent/server/app_server.py b/src/mcp_agent/server/app_server.py index 09bb3791c..ae09e3aba 100644 --- a/src/mcp_agent/server/app_server.py +++ b/src/mcp_agent/server/app_server.py @@ -2,7 +2,6 @@ MCPAgentServer - Exposes MCPApp as MCP server, and mcp-agent workflows and agents as MCP tools. """ - from __future__ import annotations import json @@ -32,7 +31,7 @@ from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse -from mcp_agent.app import MCPApp +from mcp_agent.app import MCPApp, phetch from mcp_agent.agents.agent import Agent from mcp_agent.core.context_dependent import ContextDependent from mcp_agent.executor.workflow import Workflow @@ -1523,11 +1522,14 @@ async def _internal_human_prompts(request: Request): except Exception: pass else: + if "icons" not in kwargs and app._icons: + kwargs["icons"] = app._icons if "auth" not in kwargs and effective_auth_settings is not None: kwargs["auth"] = effective_auth_settings if "token_verifier" not in kwargs and token_verifier is not None: kwargs["token_verifier"] = token_verifier owns_token_verifier = True + mcp = FastMCP( name=app.name or "mcp_agent_server", # TODO: saqadri (MAC) - create a much more detailed description @@ -1567,7 +1569,7 @@ async def _set_level( # region Workflow Tools - @mcp.tool(name="workflows-list") + @mcp.tool(name="workflows-list", icons=[phetch]) def list_workflows(ctx: MCPContext) -> Dict[str, Dict[str, Any]]: """ List all available workflow types with their detailed information. @@ -1610,7 +1612,7 @@ def list_workflows(ctx: MCPContext) -> Dict[str, Dict[str, Any]]: return result - @mcp.tool(name="workflows-runs-list") + @mcp.tool(name="workflows-runs-list", icons=[phetch]) async def list_workflow_runs( ctx: MCPContext, limit: int = 100, @@ -1667,7 +1669,7 @@ async def list_workflow_runs( return workflow_statuses - @mcp.tool(name="workflows-run") + @mcp.tool(name="workflows-run", icons=[phetch]) async def run_workflow( ctx: MCPContext, workflow_name: str, @@ -1694,7 +1696,7 @@ async def run_workflow( pass return await _workflow_run(ctx, workflow_name, run_parameters, **kwargs) - @mcp.tool(name="workflows-get_status") + @mcp.tool(name="workflows-get_status", icons=[phetch]) async def get_workflow_status( ctx: MCPContext, run_id: str | None = None, @@ -1740,7 +1742,7 @@ async def get_workflow_status( pass return await _workflow_status(ctx, run_id=run_id, workflow_id=workflow_id) - @mcp.tool(name="workflows-resume") + @mcp.tool(name="workflows-resume", icons=[phetch]) async def resume_workflow( ctx: MCPContext, run_id: str | None = None, @@ -1820,7 +1822,7 @@ async def resume_workflow( return result - @mcp.tool(name="workflows-cancel") + @mcp.tool(name="workflows-cancel", icons=[phetch]) async def cancel_workflow( ctx: MCPContext, run_id: str | None = None, workflow_id: str | None = None ) -> bool: @@ -2202,6 +2204,7 @@ async def _await_task(task: asyncio.Task): annotations = decl.get("annotations") icons = decl.get("icons") _meta = decl.get("meta") + # Bind per-iteration values to avoid late-binding closure bugs name_local = name wname_local = workflow_name @@ -2438,6 +2441,7 @@ def _schema_fn_proxy(*args, **kwargs): @mcp.tool( name=f"workflows-{workflow_name}-run", + icons=[phetch], description=f""" Run the '{workflow_name}' workflow and get a dict with workflow_id and run_id back. Workflow Description: {workflow_cls.__doc__} diff --git a/tests/server/test_tool_decorators.py b/tests/server/test_tool_decorators.py index 40106becb..66179b318 100644 --- a/tests/server/test_tool_decorators.py +++ b/tests/server/test_tool_decorators.py @@ -3,7 +3,7 @@ import pytest -from mcp_agent.app import MCPApp +from mcp_agent.app import MCPApp, phetch from mcp_agent.core.context import Context from mcp.types import ToolAnnotations, Icon from mcp.server.fastmcp import Context as FastMCPContext @@ -40,6 +40,7 @@ def add_tool( annotations=None, structured_output=None, meta=None, + icons=None, **kwargs, ): entry = { @@ -50,6 +51,7 @@ def add_tool( "annotations": annotations, "structured_output": structured_output, "meta": meta, + "icons": icons, } entry.update(kwargs) self.added_tools.append(entry) @@ -340,42 +342,145 @@ async def binding_tool( captured["fastmcp"] = app_ctx.fastmcp return f"done:{value}" + +@pytest.mark.asyncio +async def test_tool_decorator_defaults_to_phetch_icon_when_no_icons_provided(): + """Verify that when no icons parameter is provided, the default phetch icon is used.""" + app = MCPApp(name="test_default_icon") + await app.initialize() + + # Register a tool without specifying icons + @app.tool(name="no_icon_tool", description="Tool without icons") + async def no_icon_tool(text: str) -> str: + return text + + mcp = _ToolRecorder() server_context = type( "SC", (), {"workflows": app.workflows, "context": app.context} )() - ctx = _make_ctx(server_context) - # Simulate FastMCP attaching the app to its server for lookup paths - ctx.fastmcp._mcp_agent_app = app # type: ignore[attr-defined] + create_workflow_tools(mcp, server_context) + create_declared_function_tools(mcp, server_context) - run_info = await _workflow_run(ctx, "binding_tool", {"value": 7}) - run_id = run_info["run_id"] + # Find the registered tool and check its icons + tool_entry = next( + (entry for entry in mcp.added_tools if entry["name"] == "no_icon_tool"), None + ) + assert tool_entry is not None, "Tool should be registered" - # Workflow should have the FastMCP request context attached - workflow = await app.context.workflow_registry.get_workflow(run_id) - assert getattr(workflow, "_mcp_request_context", None) is ctx + # Extract icons from the tool entry + icons = tool_entry["icons"] + assert icons is not None, "Icons should not be None" + assert len(icons) == 1, "Should have exactly one icon" + assert icons[0] == phetch, "Icon should be the default phetch icon" - # Wait for completion so the tool function executes - for _ in range(200): - status = await app.context.workflow_registry.get_workflow_status(run_id) - if status.get("completed"): - break - await asyncio.sleep(0.01) - assert status.get("completed") is True - bound_app_ctx = getattr(ctx, "bound_app_context", None) - assert bound_app_ctx is not None - # The tool received the per-request bound context - assert captured.get("app_ctx") is bound_app_ctx - # FastMCP context argument should be the original request context - assert captured.get("ctx") is ctx - assert getattr(captured.get("ctx"), "bound_app_context", None) is bound_app_ctx - assert bound_app_ctx is not app.context - # Upstream session should be preserved on the bound context - assert bound_app_ctx.upstream_session is sentinel_session - assert captured.get("session_property") is sentinel_session - # FastMCP instance and request context bridge through the bound context - assert captured.get("fastmcp") is ctx.fastmcp - assert captured.get("request_context") is ctx.request_context - # Accessing session on the bound context should prefer upstream_session - assert bound_app_ctx.session is sentinel_session +@pytest.mark.asyncio +async def test_tool_decorator_uses_custom_icons_when_provided(): + """Verify that when icons parameter is provided, those icons are used instead of the default.""" + app = MCPApp(name="test_custom_icon") + await app.initialize() + + # Create a custom icon + custom_icon = Icon(src="data:image/png;base64,customdata") + + # Register a tool with custom icons + @app.tool( + name="custom_icon_tool", description="Tool with custom icon", icons=[custom_icon] + ) + async def custom_icon_tool(text: str) -> str: + return text + + mcp = _ToolRecorder() + server_context = type( + "SC", (), {"workflows": app.workflows, "context": app.context} + )() + + create_workflow_tools(mcp, server_context) + create_declared_function_tools(mcp, server_context) + + # Find the registered tool and check its icons + tool_entry = next( + (entry for entry in mcp.added_tools if entry["name"] == "custom_icon_tool"), None + ) + assert tool_entry is not None, "Tool should be registered" + + # Extract icons from the tool entry + icons = tool_entry["icons"] + assert icons is not None, "Icons should not be None" + assert len(icons) == 1, "Should have exactly one icon" + assert icons[0] == custom_icon, "Icon should be the custom icon, not phetch" + assert icons[0] != phetch, "Icon should NOT be the default phetch icon" + + +@pytest.mark.asyncio +async def test_async_tool_decorator_defaults_to_phetch_icon_when_no_icons_provided(): + """Verify that @app.async_tool defaults to phetch icon when no icons are provided.""" + app = MCPApp(name="test_async_default_icon") + await app.initialize() + + # Register an async tool without specifying icons + @app.async_tool(name="no_icon_async_tool", description="Async tool without icons") + async def no_icon_async_tool(text: str) -> str: + return text + + mcp = _ToolRecorder() + server_context = type( + "SC", (), {"workflows": app.workflows, "context": app.context} + )() + + create_workflow_tools(mcp, server_context) + create_declared_function_tools(mcp, server_context) + + # Find the registered tool and check its icons + tool_entry = next( + (entry for entry in mcp.added_tools if entry["name"] == "no_icon_async_tool"), None + ) + assert tool_entry is not None, "Tool should be registered" + + # Extract icons from the tool entry + icons = tool_entry["icons"] + assert icons is not None, "Icons should not be None" + assert len(icons) == 1, "Should have exactly one icon" + assert icons[0] == phetch, "Icon should be the default phetch icon" + + +@pytest.mark.asyncio +async def test_async_tool_decorator_uses_custom_icons_when_provided(): + """Verify that @app.async_tool uses custom icons when provided.""" + app = MCPApp(name="test_async_custom_icon") + await app.initialize() + + # Create a custom icon + custom_icon = Icon(src="data:image/png;base64,customasyncdata") + + # Register an async tool with custom icons + @app.async_tool( + name="custom_icon_async_tool", + description="Async tool with custom icon", + icons=[custom_icon], + ) + async def custom_icon_async_tool(text: str) -> str: + return text + + mcp = _ToolRecorder() + server_context = type( + "SC", (), {"workflows": app.workflows, "context": app.context} + )() + + create_workflow_tools(mcp, server_context) + create_declared_function_tools(mcp, server_context) + + # Find the registered tool and check its icons + tool_entry = next( + (entry for entry in mcp.added_tools if entry["name"] == "custom_icon_async_tool"), + None, + ) + assert tool_entry is not None, "Tool should be registered" + + # Extract icons from the tool entry + icons = tool_entry["icons"] + assert icons is not None, "Icons should not be None" + assert len(icons) == 1, "Should have exactly one icon" + assert icons[0] == custom_icon, "Icon should be the custom icon, not phetch" + assert icons[0] != phetch, "Icon should NOT be the default phetch icon"