Add typed workflow introspection to WorkflowClient (#739)

## Summary

- add typed `WorkflowClient` methods for workflow input/output schemas,
event schemas, and graph representations
- expose the corresponding response models from the public client
package
- cover successful responses and missing-workflow errors with client
integration tests
- add a patch changeset for `llama-agents-client`

## Validation

- `uv run pytest packages/llama-agents-client/tests -n0` (59 passed)
- `uv run ruff format --check
packages/llama-agents-client/src/llama_agents/client
packages/llama-agents-client/tests`
- `uv run ruff check
packages/llama-agents-client/src/llama_agents/client
packages/llama-agents-client/tests`
- `uv run ty check packages/llama-agents-client/src/llama_agents/client
packages/llama-agents-client/tests`
- `git diff --check origin/main...HEAD`

## Known gaps

The full-repository pre-commit run could not complete locally because
the workspace environment does not include optional AgentCore and
llamactl dependencies. The checks scoped to the affected client package
pass.

.changeset/workflow-client-introspection.md

@@ -0,0 +1,5 @@
+---
+"llama-agents-client": patch
+---
+
+Expose workflow schemas, event schemas, and graphs through WorkflowClient

packages/llama-agents-client/src/llama_agents/client/__init__.py

@@ -4,6 +4,9 @@ from .protocol import (
     HandlerData,
     HandlersListResponse,
     SendEventResponse,
+    WorkflowEventsListResponse,
+    WorkflowGraphResponse,
+    WorkflowSchemaResponse,
 )
 from .protocol.serializable_events import EventEnvelopeWithMetadata
 
@@ -14,5 +17,8 @@ __all__ = [
     "HandlerData",
     "HandlersListResponse",
     "SendEventResponse",
+    "WorkflowEventsListResponse",
+    "WorkflowGraphResponse",
+    "WorkflowSchemaResponse",
     "WorkflowClient",
 ]

packages/llama-agents-client/src/llama_agents/client/client.py

@@ -25,6 +25,9 @@ from .protocol import (
     HealthResponse,
     SendEventResponse,
     Status,
+    WorkflowEventsListResponse,
+    WorkflowGraphResponse,
+    WorkflowSchemaResponse,
     WorkflowsListResponse,
 )
 from .protocol.serializable_events import (
@@ -219,6 +222,29 @@ class WorkflowClient:
 
             return WorkflowsListResponse.model_validate(response.json())
 
+    async def get_workflow_schema(self, workflow_name: str) -> WorkflowSchemaResponse:
+        """Get the start and stop event schemas for a workflow."""
+        async with self._get_client() as client:
+            response = await client.get(f"/workflows/{workflow_name}/schema")
+            _raise_for_status_with_body(response)
+            return WorkflowSchemaResponse.model_validate(response.json())
+
+    async def get_workflow_events_schema(
+        self, workflow_name: str
+    ) -> WorkflowEventsListResponse:
+        """Get the schemas of events registered for a workflow."""
+        async with self._get_client() as client:
+            response = await client.get(f"/workflows/{workflow_name}/events")
+            _raise_for_status_with_body(response)
+            return WorkflowEventsListResponse.model_validate(response.json())
+
+    async def get_workflow_graph(self, workflow_name: str) -> WorkflowGraphResponse:
+        """Get the directed graph representation of a workflow."""
+        async with self._get_client() as client:
+            response = await client.get(f"/workflows/{workflow_name}/representation")
+            _raise_for_status_with_body(response)
+            return WorkflowGraphResponse.model_validate(response.json())
+
     async def run_workflow(
         self,
         workflow_name: str,

packages/llama-agents-client/tests/client/test_client.py

@@ -55,6 +55,51 @@ async def test_list_workflows(client: WorkflowClient) -> None:
     assert "crashing" in wfs.workflows
 
 
+@pytest.mark.asyncio
+async def test_get_workflow_schema(client: WorkflowClient) -> None:
+    schema = await client.get_workflow_schema("greeting")
+
+    assert schema.start["title"] == "InputEvent"
+    assert schema.stop["title"] == "OutputEvent"
+
+
+@pytest.mark.asyncio
+async def test_get_workflow_events_schema(client: WorkflowClient) -> None:
+    response = await client.get_workflow_events_schema("greeting")
+
+    event_titles = {event["title"] for event in response.events}
+    assert {"InputEvent", "OutputEvent"} <= event_titles
+
+
+@pytest.mark.asyncio
+async def test_get_workflow_graph(client: WorkflowClient) -> None:
+    response = await client.get_workflow_graph("greeting")
+
+    assert response.graph.name == "GreetingWorkflow"
+    assert {node.id for node in response.graph.nodes} >= {
+        "first_step",
+        "second_step",
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "method_name",
+    [
+        "get_workflow_schema",
+        "get_workflow_events_schema",
+        "get_workflow_graph",
+    ],
+)
+async def test_workflow_introspection_unknown_workflow_raises(
+    client: WorkflowClient, method_name: str
+) -> None:
+    method = getattr(client, method_name)
+
+    with pytest.raises(httpx.HTTPStatusError, match="404 Not Found"):
+        await method("missing")
+
+
 @pytest.mark.asyncio
 async def test_run_nowait_and_stream_events(client: WorkflowClient) -> None:
     handler = await client.run_workflow_nowait(