Guides

Building a workflow

A workflow is a graph: its steps are stated in edges rather than chosen by a model, so the path is reviewable, testable and the same every time. What a graph is and when to prefer one over an agent is in agents and workflows; this page is how to write one.

elan new workflow guest-check-in

That writes workflows/guest_check_in/workflow.py and declares the slug in elan.yaml. Slugs share one namespace with agents and follow the same rules — see the two halves.

The shape

from agentkit_core.manifest import Billing, Limits, WorkflowInput, WorkflowManifest
from google.adk.events.event import Event
from google.adk.workflow import DEFAULT_ROUTE, START, FunctionNode, Workflow
from google.genai import types

MANIFEST = WorkflowManifest(
    slug="guest-check-in",
    version="0.1.0",
    display_name="Guest Check-in",
    description="Confirms a booking and assigns a room.",
    limits=Limits(max_llm_calls=6, timeout_s=120),
    billing=Billing(sku="workflow.planning.standard"),
    max_concurrency=1,
    inputs=[
        WorkflowInput(name="guest_name", label="Guest name",
                      help="As it appears on the reservation."),
        WorkflowInput(name="nights", label="Nights", type="integer", default=1),
    ],
)


def confirm_booking(node_input, guest_name: str = "", nights: int = 1):
    """An authority decision, so code and not a model."""
    if guest_name.lower() != "ada lovelace":
        return Event(route=["NOT_FOUND"], output={"guest_name": guest_name})
    return Event(route=["CONFIRMED"],
                 output={"guest_name": guest_name, "nights": nights})


def assign_room(node_input: dict) -> dict:
    return {**node_input, "room": "712"}


def report(node_input: dict):
    # The LAST node: a plain return would reach a caller as nothing.
    return Event(content=types.Content(role="model", parts=[types.Part(
        text=f"Checked in: room {node_input['room']}.")]))


def decline(node_input):
    return Event(content=types.Content(role="model", parts=[types.Part(
        text="No reservation matches that name.")]))


def build(ctx) -> Workflow:
    confirm = FunctionNode(name="confirm_booking", func=confirm_booking)
    assign = FunctionNode(name="assign_room", func=assign_room)
    reported = FunctionNode(name="report", func=report)
    declined = FunctionNode(name="decline", func=decline)
    return Workflow(
        name="guest_check_in",
        state_schema=ctx.state_schema(),
        edges=[
            # A dict is a routing map; DEFAULT_ROUTE catches anything unnamed.
            (START, confirm, {"CONFIRMED": assign, DEFAULT_ROUTE: declined}),
            (assign, reported),
        ],
    )

The class is google.adk.Workflow, also importable from google.adk.workflow with the node types. ADK's public documentation describes a GraphAgent API that does not ship; when the two disagree, believe the installed package.

MANIFEST

Everything in an agent's manifest applies except model, tools and compaction, which a WorkflowManifest does not have. Models belong to the LlmAgent nodes inside the graph, each built with ctx.model. A workflow adds:

nodes takes a NodeOverrides per node — the operational knobs live in the manifest, where they are versioned, while the topology stays in code:

Declared inputs and the state schema

inputs is three things written once: the console's launch form, the values a caller may send, and the graph's own schema. Pass it to the graph as Workflow(state_schema=ctx.state_schema()), and ADK refuses to construct a graph whose function node asks for a parameter the manifest does not declare — a build-time error rather than a failure halfway through a run.

It is not optional once inputs is non-empty. A graph built without it fails the pod at start, saying so. Attaching a schema after construction would do nothing — ADK validates against the schema the graph was built with — so the check is that it was there from the start.

inputs=[
    WorkflowInput(name="brief", label="What is the event?", type="text"),
    WorkflowInput(name="headcount", label="Expected guests",
                  type="integer", default=100),
    WorkflowInput(name="event_date", label="Target date", type="date",
                  required=False, default=""),
]

Each input takes name, label, type (default string), required (default True), default, help and, for an enum, options as InputOption(value=..., label=...).

A type is a widget hint as much as a type, and three of them are strings by the time a node sees them:

Nodes and edges

A FunctionNode runs a Python function; an LlmAgent placed in the edges runs a model and takes its node name from the agent. A JoinNode waits for every branch that feeds it.

What a function node receives

• node_input is the previous node's output. For the first node it is the message the run was started with.
• ctx, if the function asks for it, is ADK's node context — ctx.state among it.
• Every other parameter is looked up in session state by name, which is where declared inputs land. A parameter missing from state takes its Python default; with no default, ADK raises Missing value for parameter. Give state-bound parameters defaults and a run launched without a form — or called as a tool — still works.

Edges

An edge is a tuple read left to right, and two kinds of element inside one mean something special. A tuple is a parallel fan-out; a dict is a routing map.

edges=[
    # Three model nodes in parallel; the JoinNode waits for all three.
    (START, intake, (venue_scout, vendor_shortlist, budget_model), gather),
    (gather, compose_brief, approval_check),
    # Only an explicit AUTO_APPROVE skips the person.
    (approval_check, {"AUTO_APPROVE": finalize, DEFAULT_ROUTE: approval_gate}),
    (approval_gate, finalize),
]

A node chooses a route by returning Event(route=["NAME"], output=...). Name the routes you handle and let DEFAULT_ROUTE catch the rest, so an unrecognized route declines or escalates rather than doing the risky thing by accident. The route lands on event.actions.route; reading event.route raises rather than returning None.

Pausing and resumability

A node pauses for a person by yielding a RequestInput:

from google.adk.events.request_input import RequestInput


def approval_gate(node_input):
    estimate = node_input.get("estimate") if isinstance(node_input, dict) else None
    yield RequestInput(message=f"This plan is estimated at {estimate}. Approve to continue.",
                       payload={"estimate": estimate})

Declare resumable=True on any graph that waits — for a person, or for a long-running tool. It turns on ADK's resumability, which records each node's status as the run goes so a paused run can carry on from where it stopped. The run stops at the pause and nothing is held while it waits; the answer arrives later as its own delivery.

• A resume replays the prefix without re-running it — no extra model calls. ADK documents resumption as at-least-once, though, so a node with side effects should be idempotent on something stable: the invocation id does not change across a resume.
• Without resumable, ADK writes no per-node status checkpoint, so the console's run view infers each step's status from its events instead.

Invoking a workflow

The same route, the same ordering and the same events as an agent. A workflow is a target like any other, named in target_slug; there is no separate endpoint. The route's full reference is on invocations.

curl -X POST "https://api.openelan.ai/v1/invocations" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "target_slug": "event-planner",
    "project": "agents",
    "session_id": "web-9a3f1b2c4d5e",
    "text": "offsite for the platform team",
    "inputs": {"brief": "two-day offsite, London", "headcount": 24}
  }'

text is still required and non-empty, even when everything you care about is in inputs. It is the message the first node receives as node_input, and an empty one is a 422, not a launch.

inputs, and the trap in it

inputs carries the declared launch inputs, not arbitrary data — and a partly filled inputs is stricter than none at all, so adding a key can break a call that worked without it.

So {} launches event-planner happily and {"headcount": 24} fails it, because brief is required, has no default, and you have declared yourself to be filling in the form.

A bad inputs is an error event, not a 400. The API has no copy of your manifest — it lives in your image — so it passes inputs through and answers 202; the pod validates, and the refusal arrives on the stream naming the field:

{"kind": "error", "payload": {"error": "event-planner: no declared input named 'guests'", "type": "ValueError"}}

Three ways to earn one: a key the manifest does not declare, a required input left empty once you are filling the form, and a value that will not convert. Values are converted, so strings from a form work: {"headcount": "24"} is fine for an integer, and "two dozen" is the third error.

inputs is ignored entirely for an agent, and for a workflow that declares none — silently, not refused. If values are not reaching the graph, check that the manifest declares them before looking anywhere else.

Reading a workflow's events

The envelope is an agent's, with three differences that matter:

• target_kind is workflow. Keep it apart from agent in anything you measure: a workflow's duration can include waiting for a person, which makes a shared latency percentile meaningless.
• Node identity is payload.node_info.path, not payload.author. Only LlmAgent nodes set an author, so grouping by it folds every function node into one unnamed bucket. A path reads guest_check_in@1/confirm_booking@1.
• A function node contributes no text. Its event has an output and no content, so text extraction correctly finds nothing — a silent workflow is usually its last node returning a plain value.

Resuming one

A paused run emits two paused events and no finished at all — the pause is where that delivery ends. The second is the one carrying interrupt_id and the node's prompt, and other events can arrive between the two. A paused run waits indefinitely, so do not hold a socket open for it: keep the invocation id, and GET /v1/runs?status=paused finds what is waiting on you. Answer a node's input with:

POST /v1/invocations/{invocation_id}/resume
{"approved": true, "note": "", "payload": {"cost_centre": "CC-3317"}}

• The body becomes the paused node's output and flows to the next node as its node_input: your payload keys plus approved, note and decided_by, which the platform sets and a payload cannot override. So a person can hand the graph a corrected address or a chosen option, not merely consent.
• A rejection does not stop the graph. approved: false arrives in the next node's input like any other value, and that node decides what no means.
• Only the person who started the run, or an admin, may answer it — anybody else is a 403. A run that is not paused is a 409. Unknown fields in the body are refused.
• A resumed turn reserves its own deposit, so a balance that ran out in between refuses it even though the first half succeeded.

A confirmation a tool raised is answered through the actions API instead; see approvals.

Calling a workflow from an agent

def build(ctx):
    return LlmAgent(
        name="hotel_concierge",
        model=ctx.model,
        instruction=INSTRUCTION,
        tools=[ctx.workflow_tool("guest-check-in")],
    )

ctx.workflow_tool(slug) builds the named workflow from its own manifest and offers it as a tool whose parameters are the workflow's declared inputs and whose description is its description. The workflow must be in the same project — one image holds them all — and must declare inputs, or the tool has no parameters to offer and building it fails. ADK's own AgentTool cannot wrap a workflow: it probes sub_agents, which a graph does not have, and raises — once when the tool is declared and again when the model calls it.

• The arguments arrive as one JSON document in the first node's node_input, not as session state. A graph that serves both a launch form and a tool call should read its declared inputs from state and fall back to parsing node_input.
• No plugin callback fires inside the nested run — not the platform's, not yours. A tier governs the workflow tool at its boundary, not each tool inside it, and the graph's tokens never reach the ledger; price the workflow tool in the caller's tool_costs to charge for it.
• The nested run is invisible on the caller's stream. The caller sees one tool call and its result — the text of the last content the graph produced.
• It runs in a throwaway in-memory session, discarded when the call returns, so a graph that pauses for a person is not one to offer as a tool.