fix(a2a): Support to_a2a(Workflow) and reject non-agent root nodes

### Link to Issue
- Related: #5487

**Problem:**
`to_a2a(workflow)` fails because `AgentCardBuilder` requires field `sub_agents` which `Workflow` does not have.

**Solution:**
- Make `to_a2a()` and `AgentCardBuilder` accept `Workflow` (the v2 graph orchestrator) as a root, not just `BaseAgent`. Previously crashed with`RuntimeError: 'Workflow' object has no attribute 'sub_agents'`.
- Tighten the public type contract to `BaseAgent | Workflow` and reject other `BaseNode` subtypes (e.g. `FunctionNode`, `JoinNode`) at call time with `TypeError`. They previously produced a degenerate "custom agent" card silently.

### Testing Plan
**tests/unittests/a2a/utils/test_agent_card_builder.py** — 9 new tests:
- test_get_agent_type_workflow — returns 'graph_workflow' for the new v2 Workflow.
- test_get_agent_skill_name_workflow — returns 'workflow' for Workflow.
- test_init_rejects_function_node — AgentCardBuilder(agent=FunctionNode(...)) raises TypeError (regression coverage for the runtime guard).
- test_init_rejects_arbitrary_object — AgentCardBuilder(agent="...") raises TypeError.
- test_build_succeeds_for_llm_agent — regression coverage that the original BaseAgent path still works after the type narrowing.
- test_build_succeeds_for_workflow_with_llm_agent_node — exact OP repro shape, end-to-end through build().
- test_build_succeeds_for_workflow_with_output_schema_node — covers the output_schema shape mentioned in the issue.
- test_build_succeeds_for_empty_workflow — degenerate but valid case (no edges).
- test_get_workflow_description_workflow_with_nodes — verifies graph nodes appear in the description string.
- test_get_workflow_description_empty_workflow — returns None when no nodes.
**tests/unittests/a2a/utils/test_agent_to_a2a.py** — 2 new tests, 1 rewritten:
- test_to_a2a_succeeds_for_workflow (new) — end-to-end through the actual Starlette lifespan (the exact code path that crashed in the OP's repro).
- test_to_a2a_rejects_function_node (new) — to_a2a(FunctionNode(...)) raises TypeError at call time.
- test_to_a2a_rejects_non_agent_non_workflow (rewrote existing test_to_a2a_with_invalid_agent_type) — now asserts TypeError raised eagerly at to_a2a() call time instead of AttributeError raised lazily during request handling. This is a deliberate behavior change, not a regression: the old test encoded buggy lazy-failure UX where misuse only surfaced when a client hit the endpoint.

**Unit Tests:**

- [x] I have added or updated unit tests for my change.
- [x] All unit tests pass locally.

_Please include a summary of passed `pytest` results._
ran `uv run pytest tests/unittests`

`6275 passed, 14 skipped, 25 xfailed, 10 xpassed, 2532 warnings `

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document.
- [x] I have performed a self-review of my own code.
- [x] I have commented my code, particularly in hard-to-understand areas.
- [x] I have added tests that prove my fix is effective or that my feature works.
- [x] New and existing unit tests pass locally with my changes.
- [x] I have manually tested my changes end-to-end.
- [x] Any dependent changes have been merged and published in downstream modules.

Merge #5710

Change-Id: I2330b2f3036cfd99c89b5d5c492f1c46c275974f

Author: sokoliva

src/google/adk/a2a/utils/agent_card_builder.py

@@ -32,24 +32,27 @@
 from ...agents.parallel_agent import ParallelAgent
 from ...agents.sequential_agent import SequentialAgent
 from ...tools.example_tool import ExampleTool
+from ...workflow._base_node import BaseNode
+from ...workflow._base_node import START
+from ...workflow._workflow import Workflow
 from ..experimental import a2a_experimental
 
 logger = logging.getLogger('google_adk.' + __name__)
 
 
 @a2a_experimental
 class AgentCardBuilder:
-  """Builder class for creating agent cards from ADK agents.
+  """Builder class for creating agent cards from ADK agents or workflows.
 
-  This class provides functionality to convert ADK agents into A2A agent cards,
-  including extracting skills, capabilities, and metadata from various agent
-  types.
+  This class provides functionality to convert an ADK BaseAgent (e.g. LlmAgent)
+  or a Workflow into an A2A agent card, including extracting skills,
+  capabilities, and metadata.
   """
 
   def __init__(
       self,
       *,
-      agent: BaseAgent,
+      agent: BaseAgent | Workflow,
       rpc_url: Optional[str] = None,
       capabilities: Optional[AgentCapabilities] = None,
       doc_url: Optional[str] = None,
@@ -59,6 +62,11 @@ def __init__(
   ):
     if not agent:
       raise ValueError('Agent cannot be None or empty.')
+    if not isinstance(agent, (BaseAgent, Workflow)):
+      raise TypeError(
+          'AgentCardBuilder requires a BaseAgent or Workflow, got '
+          f'{type(agent).__name__}.'
+      )
 
     self._agent = agent
     self._rpc_url = rpc_url or 'http://localhost:80/a2a'
@@ -96,8 +104,17 @@ async def build(self) -> AgentCard:
 
 
 # Module-level helper functions
-async def _build_primary_skills(agent: BaseAgent) -> List[AgentSkill]:
-  """Build skills for any agent type."""
+def _iter_child_nodes(agent: BaseNode) -> List[BaseNode]:
+  """Returns the immediate child nodes of an agent or a workflow."""
+  if isinstance(agent, BaseAgent):
+    return list(agent.sub_agents)
+  if isinstance(agent, Workflow) and agent.graph is not None:
+    return [n for n in agent.graph.nodes if n.name != START.name]
+  return []
+
+
+async def _build_primary_skills(agent: BaseNode) -> List[AgentSkill]:
+  """Build skills for any node type."""
   if isinstance(agent, LlmAgent):
     return await _build_llm_agent_skills(agent)
   else:
@@ -140,10 +157,10 @@ async def _build_llm_agent_skills(agent: LlmAgent) -> List[AgentSkill]:
   return skills
 
 
-async def _build_sub_agent_skills(agent: BaseAgent) -> List[AgentSkill]:
-  """Build skills for all sub-agents."""
+async def _build_sub_agent_skills(agent: BaseNode) -> List[AgentSkill]:
+  """Build skills for all child nodes (sub-agents or workflow nodes)."""
   sub_agent_skills = []
-  for sub_agent in agent.sub_agents:
+  for sub_agent in _iter_child_nodes(agent):
     try:
       sub_skills = await _build_primary_skills(sub_agent)
       for skill in sub_skills:
@@ -225,8 +242,8 @@ def _build_code_executor_skill(agent: LlmAgent) -> AgentSkill:
   )
 
 
-async def _build_non_llm_agent_skills(agent: BaseAgent) -> List[AgentSkill]:
-  """Build skills for non-LLM agents."""
+async def _build_non_llm_agent_skills(agent: BaseNode) -> List[AgentSkill]:
+  """Build skills for non-LLM agents and workflow nodes."""
   skills = []
 
   # 1. Agent skill (main agent skill)
@@ -249,8 +266,8 @@ async def _build_non_llm_agent_skills(agent: BaseAgent) -> List[AgentSkill]:
       )
   )
 
-  # 2. Sub-agent orchestration skill (for agents with sub-agents)
-  if agent.sub_agents:
+  # 2. Orchestration skill (for agents/workflows with child nodes)
+  if _iter_child_nodes(agent):
     orchestration_skill = _build_orchestration_skill(agent, agent_type)
     if orchestration_skill:
       skills.append(orchestration_skill)
@@ -259,11 +276,11 @@ async def _build_non_llm_agent_skills(agent: BaseAgent) -> List[AgentSkill]:
 
 
 def _build_orchestration_skill(
-    agent: BaseAgent, agent_type: str
+    agent: BaseNode, agent_type: str
 ) -> Optional[AgentSkill]:
-  """Build orchestration skill for agents with sub-agents."""
+  """Build orchestration skill for agents/workflows with child nodes."""
   sub_agent_descriptions = []
-  for sub_agent in agent.sub_agents:
+  for sub_agent in _iter_child_nodes(agent):
     description = sub_agent.description or 'No description'
     sub_agent_descriptions.append(f'{sub_agent.name}: {description}')
 
@@ -281,7 +298,7 @@ def _build_orchestration_skill(
   )
 
 
-def _get_agent_type(agent: BaseAgent) -> str:
+def _get_agent_type(agent: BaseNode) -> str:
   """Get the agent type for tagging."""
   if isinstance(agent, LlmAgent):
     return 'llm'
@@ -291,21 +308,23 @@ def _get_agent_type(agent: BaseAgent) -> str:
     return 'parallel_workflow'
   elif isinstance(agent, LoopAgent):
     return 'loop_workflow'
+  elif isinstance(agent, Workflow):
+    return 'graph_workflow'
   else:
     return 'custom_agent'
 
 
-def _get_agent_skill_name(agent: BaseAgent) -> str:
+def _get_agent_skill_name(agent: BaseNode) -> str:
   """Get the skill name based on agent type."""
   if isinstance(agent, LlmAgent):
     return 'model'
-  elif isinstance(agent, (SequentialAgent, ParallelAgent, LoopAgent)):
+  elif isinstance(agent, (SequentialAgent, ParallelAgent, LoopAgent, Workflow)):
     return 'workflow'
   else:
     return 'custom'
 
 
-def _build_agent_description(agent: BaseAgent) -> str:
+def _build_agent_description(agent: BaseNode) -> str:
   """Build agent description from agent.description and workflow-specific descriptions."""
   description_parts = []
 
@@ -382,9 +401,9 @@ def _replace_pronouns(text: str) -> str:
   )
 
 
-def _get_workflow_description(agent: BaseAgent) -> Optional[str]:
-  """Get workflow-specific description for non-LLM agents."""
-  if not agent.sub_agents:
+def _get_workflow_description(agent: BaseNode) -> Optional[str]:
+  """Get workflow-specific description for non-LLM agents and workflows."""
+  if not _iter_child_nodes(agent):
     return None
 
   if isinstance(agent, SequentialAgent):
@@ -393,6 +412,8 @@ def _get_workflow_description(agent: BaseAgent) -> Optional[str]:
     return _build_parallel_description(agent)
   elif isinstance(agent, LoopAgent):
     return _build_loop_description(agent)
+  elif isinstance(agent, Workflow):
+    return _build_graph_workflow_description(agent)
 
   return None
 
@@ -448,13 +469,32 @@ def _build_loop_description(agent: LoopAgent) -> str:
   )
 
 
-def _get_default_description(agent: BaseAgent) -> str:
+def _build_graph_workflow_description(workflow: Workflow) -> str:
+  """Build description for a graph-based Workflow."""
+  child_nodes = _iter_child_nodes(workflow)
+  descriptions = []
+  for node in child_nodes:
+    node_description = (
+        node.description.rstrip('.')
+        if node.description
+        else f'execute the {node.name} node'
+    )
+    descriptions.append(f'{node.name}: {node_description}')
+  return (
+      'This workflow orchestrates the following nodes: '
+      + '; '.join(descriptions)
+      + '.'
+  )
+
+
+def _get_default_description(agent: BaseNode) -> str:
   """Get default description based on agent type."""
   agent_type_descriptions = {
       LlmAgent: 'An LLM-based agent',
       SequentialAgent: 'A sequential workflow agent',
       ParallelAgent: 'A parallel workflow agent',
       LoopAgent: 'A loop workflow agent',
+      Workflow: 'A graph-based workflow agent',
   }
 
   for agent_type, description in agent_type_descriptions.items():
@@ -492,7 +532,7 @@ def _extract_inputs_from_examples(examples: Optional[list[dict]]) -> list[str]:
 
 
 async def _extract_examples_from_agent(
-    agent: BaseAgent,
+    agent: BaseNode,
 ) -> Optional[List[Dict]]:
   """Extract examples from example_tool if configured; otherwise, from agent instruction."""
   if not isinstance(agent, LlmAgent):
@@ -558,7 +598,7 @@ def _extract_examples_from_instruction(
   return examples if examples else None
 
 
-def _get_input_modes(agent: BaseAgent) -> Optional[List[str]]:
+def _get_input_modes(agent: BaseNode) -> Optional[List[str]]:
   """Get input modes based on agent model."""
   if not isinstance(agent, LlmAgent):
     return None
@@ -568,7 +608,7 @@ def _get_input_modes(agent: BaseAgent) -> Optional[List[str]]:
   return None
 
 
-def _get_output_modes(agent: BaseAgent) -> Optional[List[str]]:
+def _get_output_modes(agent: BaseNode) -> Optional[List[str]]:
   """Get output modes from Agent.generate_content_config.response_modalities."""
   if not isinstance(agent, LlmAgent):
     return None

src/google/adk/a2a/utils/agent_to_a2a.py

@@ -34,6 +34,7 @@
 from ...memory.in_memory_memory_service import InMemoryMemoryService
 from ...runners import Runner
 from ...sessions.in_memory_session_service import InMemorySessionService
+from ...workflow._workflow import Workflow
 from ..executor.a2a_agent_executor import A2aAgentExecutor
 from ..executor.config import A2aAgentExecutorConfig
 from ..experimental import a2a_experimental
@@ -77,7 +78,7 @@ def _load_agent_card(
 
 @a2a_experimental
 def to_a2a(
-    agent: BaseAgent,
+    agent: BaseAgent | Workflow,
     *,
     host: str = "localhost",
     port: int = 8000,
@@ -89,10 +90,11 @@ def to_a2a(
     lifespan: Callable[[Starlette], AsyncIterator[None]] | None = None,
     agent_executor_factory: Callable[[Runner], A2aAgentExecutor] | None = None,
 ) -> Starlette:
-  """Convert an ADK agent to a A2A Starlette application.
+  """Convert an ADK BaseAgent or Workflow to an A2A Starlette application.
 
   Args:
-      agent: The ADK agent to convert
+      agent: The ADK BaseAgent (e.g. LlmAgent) or Workflow to
+        convert.
       host: The host for the A2A RPC URL (default: "localhost")
       port: The port for the A2A RPC URL (default: 8000)
       protocol: The protocol for the A2A RPC URL (default: "http")
@@ -152,16 +154,20 @@ async def lifespan(app):
   adk_logger.setLevel(logging.INFO)
 
   def create_runner() -> Runner:
-    """Create a runner for the agent."""
-    return Runner(
-        app_name=agent.name or "adk_agent",
-        agent=agent,
+    """Create a runner for the agent or workflow."""
+    runner_kwargs = {
+        "app_name": agent.name or "adk_agent",
         # Use minimal services - in a real implementation these could be configured
-        artifact_service=InMemoryArtifactService(),
-        session_service=InMemorySessionService(),
-        memory_service=InMemoryMemoryService(),
-        credential_service=InMemoryCredentialService(),
-    )
+        "artifact_service": InMemoryArtifactService(),
+        "session_service": InMemorySessionService(),
+        "memory_service": InMemoryMemoryService(),
+        "credential_service": InMemoryCredentialService(),
+    }
+    if isinstance(agent, Workflow):
+      runner_kwargs["node"] = agent
+    else:
+      runner_kwargs["agent"] = agent
+    return Runner(**runner_kwargs)
 
   # Create A2A components
   if task_store is None: