Add AgentCard.from_states() classmethod for list[State] + initial_state usage

Author: prassanna-ravishankar

src/agentex/lib/types/agent_card.py

@@ -2,10 +2,14 @@
 
 import types
 import typing
-from typing import Any, get_args, get_origin
+from enum import Enum
+from typing import TYPE_CHECKING, Any, get_args, get_origin
 
 from pydantic import BaseModel
 
+if TYPE_CHECKING:
+    from agentex.lib.sdk.state_machine.state import State
+
 
 class LifecycleState(BaseModel):
     name: str
@@ -28,6 +32,57 @@ class AgentCard(BaseModel):
     input_types: list[str] = []
     output_schema: dict | None = None
 
+    @classmethod
+    def from_states(
+        cls,
+        initial_state: str | Enum,
+        states: list[State],
+        output_event_model: type[BaseModel] | None = None,
+        extra_input_types: list[str] | None = None,
+        queries: list[str] | None = None,
+    ) -> AgentCard:
+        """Build an AgentCard directly from a list[State] + initial_state.
+
+        Agents can share their `states` list between the StateMachine and acp.py
+        without constructing a temporary StateMachine instance.
+        """
+        lifecycle_states = [
+            LifecycleState(
+                name=state.name,
+                description=state.workflow.description,
+                waits_for_input=state.workflow.waits_for_input,
+                accepts=list(state.workflow.accepts),
+                transitions=[
+                    t.value if isinstance(t, Enum) else str(t)
+                    for t in state.workflow.transitions
+                ],
+            )
+            for state in states
+        ]
+
+        initial = initial_state.value if isinstance(initial_state, Enum) else initial_state
+
+        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 ls in lifecycle_states:
+            derived_input_types.update(ls.accepts)
+
+        return cls(
+            lifecycle=AgentLifecycle(
+                states=lifecycle_states,
+                initial_state=initial,
+                queries=queries or [],
+            ),
+            data_events=data_events,
+            input_types=sorted(derived_input_types | set(extra_input_types or [])),
+            output_schema=output_schema,
+        )
+
     @classmethod
     def from_state_machine(
         cls,
@@ -36,21 +91,37 @@ def from_state_machine(
         extra_input_types: list[str] | None = None,
         queries: list[str] | None = None,
     ) -> AgentCard:
-        lifecycle_data = state_machine.get_lifecycle()
-        lifecycle_data["queries"] = queries or []
+        """Build an AgentCard from a StateMachine instance. Delegates to from_states()."""
+        lifecycle = state_machine.get_lifecycle()
+        states_data = lifecycle["states"]
+        initial = lifecycle["initial_state"]
 
+        # Reconstruct lightweight State-like objects from the lifecycle dict
+        # so we can reuse from_states logic via the dict path
         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", []))
+        lifecycle_states = []
+        for s in states_data:
+            derived_input_types.update(s.get("accepts", []))
+            lifecycle_states.append(LifecycleState(
+                name=s["name"],
+                description=s.get("description", ""),
+                waits_for_input=s.get("waits_for_input", False),
+                accepts=s.get("accepts", []),
+                transitions=s.get("transitions", []),
+            ))
 
         return cls(
-            lifecycle=AgentLifecycle.model_validate(lifecycle_data),
+            lifecycle=AgentLifecycle(
+                states=lifecycle_states,
+                initial_state=initial,
+                queries=queries or [],
+            ),
             data_events=data_events,
             input_types=sorted(derived_input_types | set(extra_input_types or [])),
             output_schema=output_schema,

tests/lib/test_agent_card.py

@@ -197,6 +197,72 @@ def test_serialization_roundtrip(self):
         assert restored == card
 
 
+# --- AgentCard.from_states ---
+
+class TestAgentCardFromStates:
+    def test_lifecycle_derivation(self, sample_states):
+        card = AgentCard.from_states(initial_state=SampleState.WAITING, states=sample_states)
+
+        assert card.lifecycle is not None
+        assert card.lifecycle.initial_state == "waiting"
+        assert len(card.lifecycle.states) == 3
+
+    def test_initial_state_string(self, sample_states):
+        card = AgentCard.from_states(initial_state="waiting", states=sample_states)
+        assert card.lifecycle is not None
+        assert card.lifecycle.initial_state == "waiting"
+
+    def test_input_types_union(self, sample_states):
+        card = AgentCard.from_states(initial_state=SampleState.WAITING, states=sample_states)
+        assert card.input_types == ["doc_upload", "text"]
+
+    def test_extra_input_types(self, sample_states):
+        card = AgentCard.from_states(
+            initial_state=SampleState.WAITING,
+            states=sample_states,
+            extra_input_types=["admin_command"],
+        )
+        assert card.input_types == ["admin_command", "doc_upload", "text"]
+
+    def test_data_events_and_schema(self, sample_states):
+        card = AgentCard.from_states(
+            initial_state=SampleState.WAITING,
+            states=sample_states,
+            output_event_model=SampleOutputEvent,
+            queries=["get_current_state"],
+        )
+        assert card.data_events == ["plan_update", "status_change", "report_done"]
+        assert card.output_schema is not None
+        assert card.lifecycle is not None
+        assert card.lifecycle.queries == ["get_current_state"]
+
+    def test_state_fields(self, sample_states):
+        card = AgentCard.from_states(initial_state=SampleState.WAITING, states=sample_states)
+        assert card.lifecycle is not None
+        states_by_name = {s.name: s for s in card.lifecycle.states}
+
+        waiting = states_by_name["waiting"]
+        assert waiting.description == "Waiting for input"
+        assert waiting.waits_for_input is True
+        assert waiting.accepts == ["text", "doc_upload"]
+        assert waiting.transitions == ["processing"]
+
+    def test_matches_from_state_machine(self, sample_states, sample_sm):
+        """from_states and from_state_machine should produce identical cards."""
+        card_states = AgentCard.from_states(
+            initial_state=SampleState.WAITING,
+            states=sample_states,
+            output_event_model=SampleOutputEvent,
+            queries=["get_current_state"],
+        )
+        card_sm = AgentCard.from_state_machine(
+            state_machine=sample_sm,
+            output_event_model=SampleOutputEvent,
+            queries=["get_current_state"],
+        )
+        assert card_states == card_sm
+
+
 # --- AgentCard.from_state_machine ---
 
 class TestAgentCardFromStateMachine: