Add AgentCard feature for self-describing agent capabilities via registration_metadata

Author: prassanna-ravishankar

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -81,6 +81,9 @@ def __init__(self):
         # Agent info to return in healthz
         self.agent_id: str | None = None
 
+        # Optional agent card for registration metadata
+        self._agent_card: Any | None = None
+
     @classmethod
     def create(cls):
         """Create and initialize BaseACPServer instance"""
@@ -98,7 +101,7 @@ def get_lifespan_function(self):
         async def lifespan_context(app: FastAPI):  # noqa: ARG001
             env_vars = EnvironmentVariables.refresh()
             if env_vars.AGENTEX_BASE_URL:
-                await register_agent(env_vars)
+                await register_agent(env_vars, agent_card=self._agent_card)
                 self.agent_id = env_vars.AGENT_ID
             else:
                 logger.warning("AGENTEX_BASE_URL not set, skipping agent registration")

src/agentex/lib/sdk/fastacp/fastacp.py

@@ -2,7 +2,7 @@
 
 import os
 import inspect
-from typing import Literal
+from typing import Any, Literal
 from pathlib import Path
 from typing_extensions import deprecated
 
@@ -88,7 +88,10 @@ def locate_build_info_path() -> None:
 
     @staticmethod
     def create(
-        acp_type: Literal["sync", "async", "agentic"], config: BaseACPConfig | None = None, **kwargs
+        acp_type: Literal["sync", "async", "agentic"],
+        config: BaseACPConfig | None = None,
+        agent_card: Any | None = None,
+        **kwargs,
     ) -> BaseACPServer | SyncACP | AsyncBaseACP | TemporalACP:
         """Main factory method to create any ACP type
 
@@ -102,10 +105,17 @@ def create(
 
         if acp_type == "sync":
             sync_config = config if isinstance(config, SyncACPConfig) else None
-            return FastACP.create_sync_acp(sync_config, **kwargs)
+            instance = FastACP.create_sync_acp(sync_config, **kwargs)
         elif acp_type == "async" or acp_type == "agentic":
             if config is None:
                 config = AsyncACPConfig(type="base")
             if not isinstance(config, AsyncACPConfig):
                 raise ValueError("AsyncACPConfig is required for async/agentic ACP type")
-            return FastACP.create_async_acp(config, **kwargs)
+            instance = FastACP.create_async_acp(config, **kwargs)
+        else:
+            raise ValueError(f"Unknown acp_type: {acp_type}")
+
+        if agent_card is not None:
+            instance._agent_card = agent_card  # type: ignore[attr-defined]
+
+        return instance

src/agentex/lib/sdk/state_machine/__init__.py

@@ -3,4 +3,14 @@
 from .state_machine import StateMachine
 from .state_workflow import StateWorkflow
 
-__all__ = ["StateMachine", "StateWorkflow", "State", "NoOpWorkflow"]
+from agentex.lib.types.agent_card import AgentCard, AgentLifecycle, LifecycleState
+
+__all__ = [
+    "StateMachine",
+    "StateWorkflow",
+    "State",
+    "NoOpWorkflow",
+    "AgentCard",
+    "AgentLifecycle",
+    "LifecycleState",
+]

src/agentex/lib/sdk/state_machine/state_machine.py

@@ -129,6 +129,30 @@ async def reset_to_initial_state(self):
             span.output = {"output_state": self._initial_state}  # type: ignore[assignment,union-attr]
             await adk.tracing.end_span(trace_id=self._task_id, span=span)
 
+    def get_lifecycle(self) -> dict[str, Any]:
+        """Export the state machine's lifecycle as a dict suitable for AgentCard."""
+        states = []
+        for state in self._state_map.values():
+            workflow = state.workflow
+            states.append({
+                "name": state.name,
+                "description": workflow.description,
+                "waits_for_input": workflow.waits_for_input,
+                "accepts": list(workflow.accepts),
+                "transitions": [
+                    t.value if hasattr(t, "value") else str(t)
+                    for t in workflow.transitions
+                ],
+            })
+        initial = self._initial_state
+        if hasattr(initial, "value"):
+            initial = initial.value
+
+        return {
+            "states": states,
+            "initial_state": initial,
+        }
+
     def dump(self) -> dict[str, Any]:
         """
         Save the current state of the state machine to a serializable dictionary.

src/agentex/lib/sdk/state_machine/state_workflow.py

@@ -11,6 +11,11 @@
 
 
 class StateWorkflow(ABC):
+    description: str = ""
+    waits_for_input: bool = False
+    accepts: list[str] = []
+    transitions: list[str] = []
+
     @abstractmethod
     async def execute(
         self, state_machine: "StateMachine", state_machine_data: BaseModel | None = None

src/agentex/lib/types/agent_card.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import typing
+from typing import Any, get_args, get_origin
+
+from pydantic import BaseModel
+
+
+class LifecycleState(BaseModel):
+    name: str
+    description: str = ""
+    waits_for_input: bool = False
+    accepts: list[str] = []
+    transitions: list[str] = []
+
+
+class AgentLifecycle(BaseModel):
+    states: list[LifecycleState]
+    initial_state: str
+    queries: list[str] = []
+
+
+class AgentCard(BaseModel):
+    protocol: str = "acp"
+    lifecycle: AgentLifecycle | None = None
+    data_events: list[str] = []
+    input_types: list[str] = []
+    output_schema: dict | None = None
+
+    @classmethod
+    def from_state_machine(
+        cls,
+        state_machine: Any,
+        output_event_model: type[BaseModel] | None = None,
+        extra_input_types: list[str] | None = None,
+        queries: list[str] | None = None,
+    ) -> AgentCard:
+        lifecycle_data = state_machine.get_lifecycle()
+        lifecycle_data["queries"] = queries or []
+
+        data_events: list[str] = []
+        output_schema: dict | None = None
+        if output_event_model:
+            data_events = extract_literal_values(output_event_model, "type")
+            output_schema = output_event_model.model_json_schema()
+
+        derived_input_types: set[str] = set()
+        for state in lifecycle_data["states"]:
+            derived_input_types.update(state.get("accepts", []))
+
+        return cls(
+            lifecycle=AgentLifecycle.model_validate(lifecycle_data),
+            data_events=data_events,
+            input_types=sorted(derived_input_types | set(extra_input_types or [])),
+            output_schema=output_schema,
+        )
+
+
+def extract_literal_values(model: type[BaseModel], field: str) -> list[str]:
+    """Extract allowed values from a Literal[...] type annotation on a Pydantic model field."""
+    field_info = model.model_fields.get(field)
+    if field_info is None:
+        return []
+
+    annotation = field_info.annotation
+    if annotation is None:
+        return []
+
+    # Unwrap Optional (Union[X, None]) to get the inner type
+    if get_origin(annotation) is typing.Union:
+        args = [a for a in get_args(annotation) if a is not type(None)]
+        annotation = args[0] if len(args) == 1 else annotation
+
+    if get_origin(annotation) is typing.Literal:
+        return list(get_args(annotation))
+
+    return []

src/agentex/lib/utils/registration.py

@@ -31,7 +31,7 @@ def get_build_info():
     except Exception:
         return None
 
-async def register_agent(env_vars: EnvironmentVariables):
+async def register_agent(env_vars: EnvironmentVariables, agent_card=None):
     """Register this agent with the Agentex server"""
     if not env_vars.AGENTEX_BASE_URL:
         logger.warning("AGENTEX_BASE_URL is not set, skipping registration")
@@ -45,13 +45,18 @@ async def register_agent(env_vars: EnvironmentVariables):
     )
 
     # Prepare registration data
+    registration_metadata = get_build_info() or {}
+    if agent_card is not None:
+        card_data = agent_card.model_dump() if hasattr(agent_card, "model_dump") else agent_card
+        registration_metadata["agent_card"] = card_data
+
     registration_data = {
         "name": env_vars.AGENT_NAME,
         "description": description,
         "acp_url": full_acp_url,
         "acp_type": env_vars.ACP_TYPE,
         "principal_context": get_auth_principal(env_vars),
-        "registration_metadata": get_build_info()
+        "registration_metadata": registration_metadata,
     }
 
     if env_vars.AGENT_ID: