update mcp dependency; add ability to add icons (lastmile-ai#560)

* update mcp dependency; add ability to add icons

* fix linter issue

* fix test cases

* review comments / cleanup / tests

* use icon from s3; fix example icon

* fix linting issues

Author: roman-van-der-krogt

README.md

@@ -592,6 +592,36 @@ lost_baggage = SwarmAgent(
 
 </details>
 
+### 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.

examples/mcp_agent_server/temporal/mag.png

@@ -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,
 )

examples/mcp_agent_server/temporal/main.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(
                 {

src/mcp_agent/app.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__}