feat(a2a): add agent card and rpc skeleton

.env.example

@@ -32,4 +32,7 @@ UNSUBSCRIBE_AUTO_ENABLED=true
 UNSUBSCRIBE_AUTO_INTERVAL_MINUTES=720
 UNSUBSCRIBE_HTTP_TIMEOUT_SECONDS=12
 UNSUBSCRIBE_USER_AGENT=Mozilla/5.0 (compatible; PersonalAgentUnsubscribe/1.0)
+A2A_PUBLIC_BASE_URL=
+A2A_AGENT_NAME=Personal Agent
+A2A_AGENT_DESCRIPTION=Personal productivity agent for calendar availability and email operations.
 LOG_LEVEL=INFO

app/a2a/__init__.py

@@ -0,0 +1,3 @@
+from app.a2a.router import router as a2a_router
+
+__all__ = ["a2a_router"]

app/a2a/agent_card.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import Request
+
+from app.config import Settings
+
+
+def build_agent_card(settings: Settings, request: Request) -> dict[str, Any]:
+    base_url = _resolve_base_url(settings=settings, request=request)
+    return {
+        "name": settings.a2a_agent_name,
+        "description": settings.a2a_agent_description,
+        "url": f"{base_url}/a2a/rpc",
+        "version": "0.1.0",
+        "protocolVersion": "1.0",
+        "defaultInputModes": ["application/json"],
+        "defaultOutputModes": ["application/json"],
+        "capabilities": {
+            "streaming": False,
+            "pushNotifications": False,
+            "stateTransitionHistory": False,
+        },
+        "securitySchemes": {
+            "bearerAuth": {
+                "type": "http",
+                "scheme": "bearer",
+                "description": "Use the same Bearer/API key auth as the REST API.",
+            }
+        },
+        "security": [{"bearerAuth": []}],
+        "skills": [
+            {
+                "id": "availability.query",
+                "name": "Check Availability",
+                "description": "Checks Google Calendar availability for a given time range.",
+                "tags": ["calendar", "availability", "scheduling"],
+                "examples": [
+                    "Is calendar primary free from 2026-03-10T09:00:00+01:00 to 2026-03-10T10:00:00+01:00?"
+                ],
+                "inputModes": ["application/json"],
+                "outputModes": ["application/json"],
+            }
+        ],
+    }
+
+
+def _resolve_base_url(*, settings: Settings, request: Request) -> str:
+    if settings.a2a_public_base_url:
+        return settings.a2a_public_base_url.rstrip("/")
+    return str(request.base_url).rstrip("/")

app/a2a/models.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class A2ARpcRequest(BaseModel):
+    jsonrpc: str = "2.0"
+    id: str | int | None = None
+    method: str
+    params: dict[str, Any] = Field(default_factory=dict)
+
+
+class A2ARpcError(BaseModel):
+    code: int
+    message: str
+    data: dict[str, Any] | None = None
+
+
+class A2ARpcResponse(BaseModel):
+    jsonrpc: str = "2.0"
+    id: str | int | None = None
+    result: dict[str, Any] | None = None
+    error: A2ARpcError | None = None

app/a2a/router.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Request, Response
+
+from app.a2a.agent_card import build_agent_card
+from app.a2a.models import A2ARpcError, A2ARpcRequest, A2ARpcResponse
+from app.config import get_settings
+
+settings = get_settings()
+
+router = APIRouter(tags=["a2a"])
+
+
+@router.get("/.well-known/agent-card.json")
+def get_agent_card(request: Request, response: Response) -> dict[str, Any]:
+    response.headers["A2A-Version"] = "1.0"
+    return build_agent_card(settings=settings, request=request)
+
+
+@router.post("/a2a/rpc", response_model=A2ARpcResponse)
+def a2a_rpc(payload: A2ARpcRequest, response: Response) -> A2ARpcResponse:
+    response.headers["A2A-Version"] = "1.0"
+    if payload.jsonrpc != "2.0":
+        return _error_response(
+            request_id=payload.id,
+            code=-32600,
+            message="Invalid Request: jsonrpc must be '2.0'.",
+        )
+
+    if payload.method in {"ping", "health.ping", "health/ping"}:
+        return A2ARpcResponse(
+            id=payload.id,
+            result={"status": "ok", "agent": settings.a2a_agent_name},
+        )
+
+    return _error_response(
+        request_id=payload.id,
+        code=-32601,
+        message=f"Method '{payload.method}' is not implemented yet.",
+    )
+
+
+def _error_response(request_id: str | int | None, code: int, message: str) -> A2ARpcResponse:
+    return A2ARpcResponse(
+        id=request_id,
+        error=A2ARpcError(code=code, message=message),
+    )

app/config.py

@@ -42,6 +42,9 @@ class Settings:
     unsubscribe_auto_interval_minutes: int
     unsubscribe_http_timeout_seconds: float
     unsubscribe_user_agent: str
+    a2a_public_base_url: str | None
+    a2a_agent_name: str
+    a2a_agent_description: str
     log_level: str
 
 
@@ -96,6 +99,12 @@ def get_settings() -> Settings:
             "UNSUBSCRIBE_USER_AGENT",
             "Mozilla/5.0 (compatible; PersonalAgentUnsubscribe/1.0; +https://example.local)",
         ),
+        a2a_public_base_url=os.getenv("A2A_PUBLIC_BASE_URL", "").strip() or None,
+        a2a_agent_name=os.getenv("A2A_AGENT_NAME", "Personal Agent"),
+        a2a_agent_description=os.getenv(
+            "A2A_AGENT_DESCRIPTION",
+            "Personal productivity agent for calendar availability and email operations.",
+        ),
         log_level=os.getenv("LOG_LEVEL", "INFO"),
     )
 

app/main.py

@@ -10,6 +10,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from fastapi import Depends, FastAPI, Header, HTTPException, Query, status
 from pydantic import BaseModel, Field
 
+from app.a2a import a2a_router
 from app.config import get_settings
 from app.core.service import CoreAgentService
 from app.security import AuthBackend
@@ -345,6 +346,7 @@ async def lifespan(app: FastAPI):
 
 
 app = FastAPI(title="Personal Agent", version="0.3.0", lifespan=lifespan) # type: ignore
+app.include_router(a2a_router)
 
 
 @app.get("/health")