Merge pull request #103 from yuvalk/logo

feat: add icon_url to agent card

Author: luis5tb

src/lightspeed_agent/api/app.py

@@ -9,12 +9,15 @@
 """
 
 import logging
+import pathlib
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Any
 
 from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPI
+from fastapi import Request
 from fastapi.middleware.cors import CORSMiddleware
+from starlette.responses import FileResponse
 
 from lightspeed_agent.api.a2a.a2a_setup import setup_a2a_routes
 from lightspeed_agent.api.a2a.agent_card import get_agent_card_dict
@@ -24,9 +27,18 @@
 from lightspeed_agent.ratelimit import RateLimitMiddleware, get_redis_rate_limiter
 from lightspeed_agent.security import RequestBodyLimitMiddleware, SecurityHeadersMiddleware
 
+_LOGO_PATH = pathlib.Path(__file__).parent.parent / "static" / "logo.png"
+
 logger = logging.getLogger(__name__)
 
 
+def _agent_card_response(request: Request) -> dict[str, Any]:
+    """Build agent card dict with a dynamic iconUrl derived from the request base URL."""
+    card = get_agent_card_dict()
+    icon_url = f"{str(request.base_url).rstrip('/')}/static/logo.png"
+    return {**card, "iconUrl": icon_url}
+
+
 @asynccontextmanager
 async def lifespan(app: A2AFastAPI) -> AsyncIterator[None]:
     """Application lifespan manager for startup/shutdown events."""
@@ -134,19 +146,32 @@ def create_app() -> A2AFastAPI:
         lifespan=lifespan,
     )
 
+    # Serve the Red Hat logo for the agent card iconUrl
+    @app.get("/static/logo.png")
+    async def serve_logo() -> FileResponse:
+        """Serve the agent logo image."""
+        return FileResponse(_LOGO_PATH, media_type="image/png")
+
+    # Custom agent card endpoint registered BEFORE setup_a2a_routes so
+    # FastAPI's first-match routing picks it up instead of the SDK default.
+    @app.get("/.well-known/agent.json")
+    async def agent_card_with_icon(request: Request) -> dict[str, Any]:
+        """AgentCard endpoint with dynamic iconUrl."""
+        return _agent_card_response(request)
+
     # Set up A2A protocol routes using ADK's built-in integration
     # This provides:
-    # - GET /.well-known/agent.json - AgentCard
+    # - GET /.well-known/agent.json - AgentCard (overridden above)
     # - POST / - JSON-RPC 2.0 endpoint for message/send, message/stream, etc.
     # The ADK integration handles SSE streaming, task management, and
     # event conversion automatically.
     setup_a2a_routes(app)
 
     # Alias for agent card (some clients use agent-card.json)
     @app.get("/.well-known/agent-card.json")
-    async def agent_card_alias() -> dict[str, Any]:
+    async def agent_card_alias(request: Request) -> dict[str, Any]:
         """AgentCard endpoint (alias for agent.json)."""
-        return get_agent_card_dict()
+        return _agent_card_response(request)
 
     # Add authentication middleware for A2A endpoint (innermost layer)
     # Validates Red Hat SSO JWT tokens on POST / requests

src/lightspeed_agent/static/logo.png

@@ -136,6 +136,7 @@ def test_get_agent_card_dict(self):
         assert "protocolVersion" in card_dict  # aliased field
         assert "securitySchemes" in card_dict  # aliased field
         assert "defaultInputModes" in card_dict  # aliased field
+        assert "iconUrl" not in card_dict  # injected dynamically at the endpoint level
 
     def test_agent_card_all_fields(self):
         """Validate structural correctness of every AgentCard field."""
@@ -329,14 +330,29 @@ def client(self):
         rl_mod._rate_limiter = None
 
     def test_agent_card_endpoint(self, client):
-        """Test /.well-known/agent.json endpoint."""
+        """Test /.well-known/agent.json endpoint returns card with dynamic iconUrl."""
         response = client.get("/.well-known/agent.json")
 
         assert response.status_code == 200
         data = response.json()
         assert "name" in data
         assert "skills" in data
         assert "securitySchemes" in data
+        assert data["iconUrl"] == "http://testserver/static/logo.png"
+
+    def test_agent_card_alias_matches(self, client):
+        """Test /.well-known/agent-card.json returns the same card as agent.json."""
+        main = client.get("/.well-known/agent.json").json()
+        alias = client.get("/.well-known/agent-card.json").json()
+
+        assert alias["iconUrl"] == main["iconUrl"]
+
+    def test_logo_endpoint(self, client):
+        """Test GET /static/logo.png serves an image/png response."""
+        response = client.get("/static/logo.png")
+
+        assert response.status_code == 200
+        assert response.headers["content-type"] == "image/png"
 
     def test_send_message_jsonrpc(self, client):
         """Test / endpoint with JSON-RPC message/send."""