Workflow Guide

In a project workflow, the agent placed in a step runs once per step. This guide explains how to implement such an agent with the SDK's workflow runner (agenticstar_platform.workflow).

Availability

Available for workflows in the Marketplace edition 3.1 and later. Requires SDK v3.0.1 or later (the result format changed in v3.0.1, and the platform accepts only the new format). Use the latest version (v3.0.15). Agents that run in chat keep using run_marketplace_agent as before (see the Deploy Guide).

Differences from the chat runner

Item	Chat (run_marketplace_agent)	Workflow (workflow.run)
Unit of execution	One message in a conversation	One run of a workflow step
Entry point	async def my_agent(emitter, message)	async def run(context)
Input	The last user message	Settings (parameters) and input files (such as outputs of earlier steps)
Output	The function return value (body of completion_success)	Output files declared in the contract
Installation	agenticstar-platform[runner]	agenticstar-platform (core only; no config.toml or DB settings)
Environment variables the runner uses	EXECUTION_ID etc. + DB_* + WEBHOOK_URL	ASTER_RUNTIME_* and EXECUTION_ID injected by the platform
Result	completion_success / completion_failure	An outcome describing how the run ended (see below)

How it works

The runner runs as the parent process and executes your agent function in a child process. You only write the body of the function.

1. Validates the injected environment variables (ASTER_RUNTIME_PROTOCOL / ASTER_RUNTIME_BASE_URL / ASTER_RUNTIME_TOKEN / EXECUTION_ID) and fetches the execution context from the platform runtime API. If anything is invalid, it exits without calling your agent
2. Saves the start record (context.manifest audit). If it cannot confirm the save, it does not start your agent
3. Fetches the input files, verifies their digest and size, places them in /workspace/input (read-only) and copies them to /workspace/work for working. Settings files with from_parameters are generated from the parameters
4. Runs your agent function in a child process whose current directory is /workspace/work
5. After it finishes, collects, validates and saves the declared outputs and edited files, then saves the execution result (completion)

Entry point

The entry point is async def run(context). Record reads and writes to external systems in the audit before and after each operation (see Auditing external operations). Reading and writing local files is not audited.

my_agent/main.pyPython

async def run(context):
    instructions = (context.work_dir / "instructions.md").read_text(encoding="utf-8")

    inv = await context.audit.tool_invoked("get_product", effect_kind="read", target="product-api")
    product = await product_api.get(context.parameters["product_code"])  # your external API call
    await context.audit.tool_effect(inv, status="ok", count=1)

    reply = await create_reply(product, instructions)  # your processing
    (context.output_dir / "reply.md").write_text(reply, encoding="utf-8")

Only the core install is needed (pip install agenticstar-platform==3.0.15). Pass the entry point to run() in the image's start command. Your agent function runs in a child process whose current directory is /workspace/work, so make the entry-point module importable from anywhere: install it as a package, or put your application directory on PYTHONPATH.

Dockerfile (end)Dockerfile

WORKDIR /app
COPY . .
# the child process runs in /workspace/work, so make the code in /app importable
ENV PYTHONPATH=/app

CMD ["python", "-c", "from agenticstar_platform.workflow import run; run('my_agent.main:run')"]

Write CMD in exec form as above (["python", "-c", ...]) so that the runner is PID 1 of the container. A shell form (CMD python -c ..., sh -c "...") or an init process in front of the runner keeps another process holding the runner's token, which commands run by your agent can read (from v3.0.13 the runner logs a warning when PID 1 holds the token).

The exit codes of run() are:

• 0: the execution result was saved (including when the run had already ended). If your agent raised an exception or the entry point could not be imported, the exit code is still 0 as long as the result (failed/agent_error) was saved. The exit code is also 0 when the platform refused the result after a cancellation (v3.0.13+; the platform closes the run as cancelled)
• 2: the execution result could not be saved
• 3: the environment variables or the execution context are invalid (your agent was not called)

Other failures may end with a non-zero exit and the exception (the SDK does not guarantee exit codes). The platform closes a run that exits non-zero without a saved result as failed/agent_exit_nonzero, and a run that exits with 0 but has no result as failed/completion_missing after the collection grace period. Both are treated as effect unknown and sent to a human for review.

Environment variables and secrets

Your agent's process receives the following environment variables. The token the runner uses for the runtime API (ASTER_RUNTIME_TOKEN) is not passed to it. From v3.0.13 your agent's processes cannot read it from the runner's /proc/<pid>/environ either (Linux; start the image in exec form).

• Environment variables set in the agent registration of the admin console (register secrets such as LLM API keys here, as for chat agents). The names listed in the contract's fixed_env_names are passed (Fixed environment variables)
• Environment variables built from parameter values through the contract's env_bindings
• Access tokens of the connections placed in the contract's connection_slots (see Connection tokens)
• Execution correlation such as EXECUTION_ID

Reserved names such as EXECUTION_ID, ASTER_*, MP_*, PATH, PYTHON* and LD_* cannot be used in the admin console environment variables or in env_bindings.

Workflow contract

To use an agent in workflows, enable Use in workflows in the agent registration of the admin console and register a contract. The contract is stored as JSON and can also be loaded from YAML. It defines the settings users can change, the input files, the outputs and the connections.

Example contractJSON

{
  "contract_version": 1,
  "runtime_protocol": "mp-workflow/1",
  "parameters_schema": {
    "type": "object",
    "additionalProperties": false,
    "properties": {
      "product_code": {"type": "string", "maxLength": 64},
      "language": {"type": "string", "enum": ["ja", "en"]}
    },
    "required": ["product_code", "language"]
  },
  "defaults": {"product_code": "demo", "language": "ja"},
  "env_bindings": {"PRODUCT_CODE": "product_code", "REPLY_LANGUAGE": "language"},
  "files": [
    {"name": "instructions", "path": "instructions.md", "format": "markdown", "editable": true, "required": true},
    {"name": "settings", "path": "settings/agent.yaml", "format": "yaml", "from_parameters": true}
  ],
  "outputs": [
    {"name": "reply", "path": "reply.md", "format": "markdown", "required": true}
  ],
  "connection_slots": [],
  "effect_mode": "read_only",
  "limits": {"execution_timeout_seconds": 900}
}

Key	Description
runtime_protocol	"mp-workflow/1"
parameters_schema	JSON Schema of the settings users can change per step or project. type must be object, and additionalProperties must be false, including in nested object schemas. Allowed keywords: type / properties / required / additionalProperties / items / enum / minLength / maxLength / minimum / maximum / minItems / maxItems / title / description / default. Arrays require items. Up to 8 levels of nesting, 100 array elements, 64KiB for the schema and 64KiB for the parameter values (JSON)
defaults	Default parameter values, overridden by the step settings and then the project settings
env_bindings	Environment variable name → parameter name. Use it when you want to receive parameter values as environment variables. Names start with an uppercase letter and contain uppercase letters, digits and _ (up to 64 characters). Only string, number and boolean parameters; values are passed as strings (booleans as "true" / "false"). Secret-like names (containing TOKEN / SECRET / API_KEY etc.) cannot be used
files	Input files. name / path (relative path) / format (markdown / yaml / json / text / binary) / editable / required / from_parameters (yaml or json generated from all parameters) / schema (validation for yaml or json)
outputs	Outputs. name / path (relative to /workspace/output) / format / required / schema. With source: "work", the file at path relative to /workspace/work (an edited file) is collected as an output. Undeclared files are not collected
connection_slots	Connection slots: {name, allowed_provider_ids, required} (see below)
fixed_env_names	Names of the fixed environment variables passed to the step (see below). The admin console builds this from the names of the agent env, so it is not a value you write in the contract
effect_mode	read_only / idempotent_write / non_idempotent_write. Only read_only and idempotent_write runs whose effect is not unknown can be retried automatically after a failure. Agents that write externally must declare a value that matches their writes
limits.execution_timeout_seconds	Maximum execution time (seconds). 0 means no limit (up to 86400 seconds = 24 hours). Defaults to 900 seconds when omitted

To use connections, place slots in connection_slots. name starts with a letter and contains letters, digits, _ and - (up to 64 characters), and allowed_provider_ids are IDs of OAuth providers registered in the admin console. Users choose which connection to use in the project.

Example connection slotJSON

"connection_slots": [
  {"name": "gitlab", "allowed_provider_ids": ["<OAuth provider ID>"], "required": true}
]

Fixed environment variables

Configuration values that do not change per user (connection endpoints, tenant IDs, API keys of external services, and so on) are registered as the agent env in the admin console. Only the names listed in the contract's fixed_env_names are passed to the step.

• The admin console builds fixed_env_names from the names of the agent env. It is not a value you write in the contract (the values are not stored in the contract either; they are read from the admin console env at each launch)
• Therefore, adding, removing or renaming an env changes the contract, and so changes contract_digest. Certify again
• Names start with an upper-case letter and contain upper-case letters, digits and _ (up to 64 characters). Reserved names such as ASTER_ and MP_, and names already used by env_bindings, cannot be used. Unlike env_bindings, names containing TOKEN or API_KEY are allowed (because the administrator owns the values)
• An env registered as a secret is passed as a Kubernetes Secret reference, and its value is not left in the execution information

The platform database connection is not passed to steps. If you need a connection endpoint (such as the URL of an MCP server), pass it in these fixed environment variables.

In addition to the contract, an administrator registers the certification information. Only agents that have all five of the following in certification can be started in workflows.

• sdk_version — the tested SDK version
• adapter_version — the tested adapter version
• tests_ref — reference to the test results
• image_digest — digest of the certified image (sha256:<64 hex characters>)
• supported_policies — the basic rules the agent supports (an array of the ones confirmed in testing, from open, guarded and strict). If the project's basic rule is not in it, the agent does not start and the run ends as denied/policy_unenforceable

If even one is missing, the project screen shows "Unavailable (not certified)" and the agent cannot be placed in steps (Settings Guide). Runs use the image with the certified digest, so certify again after you update the image. You can look up the digest with, for example:

curl

docker buildx imagetools inspect <registry>/<image>:<tag>

Workspace and files

Location	Attribute	Purpose
/workspace/input	context.input_dir	Input files (read-only)
/workspace/work	context.work_dir	Working copy of the inputs. Edit editable files here and collect them with source: "work" in outputs
/workspace/output	context.output_dir	Where you write the declared outputs

• Limits: 20MiB per file, and 100MiB / 100 files in total per run. Files collected from the working directory (source: "work") are limited to 1MiB. Paths are 1–255 characters
• During collection, symlinks, hard links, paths pointing outside the directory and files over the limits are rejected. Optional (not required) working files that were not edited are not saved
• JSON / YAML are parsed strictly. Duplicate keys are rejected. YAML accepts a single document only, rejects anchors / aliases and explicit tags (including !!str), and does not convert yes / no or dates implicitly. Numbers with a leading zero (such as 01) are an error. Quote values you want to keep as strings

What context provides

Attribute / method	Description
execution_id / run_id / project_id / root_execution_id / entry_ref / operation_key / attempt	Execution correlation. operation_key can be used as the basis of idempotency keys for external operations
parameters	Resolved settings (read-only)
connections	Connection slot name → connection ID
effect_mode	effect_mode of the contract
input_dir / work_dir / output_dir	Workspace paths
deadlines / deadline_remaining()	Deadlines, and the seconds left until your agent is stopped (None if unknown)
audit.tool_invoked() / tool_effect() / tool_denied()	Auditing external operations (next section)
is_cancel_requested(max_age=5.0) / ensure_can_act()	Check for cancellation and deadlines
skip(reason)	Skip the step when there is nothing to process (no_target / not_applicable / nothing_to_do)
policy_unavailable()	Report that whether an operation is allowed could not be checked before any external operation
log(message)	Write to the runner log (up to 2000 characters)
connection_token(slot, min_ttl=300.0)	Refresh the access token of a connection (v3.0.5+)
gate_decision	The review decision that started this run (read-only; None for any other run; v3.0.9+)

Context cannot decide a gate (human review) or open arbitrary database connections. You can read a decision that has been made through gate_decision (see Runs that follow a review decision).

Auditing external operations

Record reads and writes to external systems in the audit before and after each operation.

1. Before the operation, call tool_invoked(tool, effect_kind=..., action_ref=..., target=...). It returns only after the record has been saved. If it cannot be saved (including after cancellation or the deadline), it raises AuditNotPersisted; do not proceed with that operation
2. After the operation, call tool_effect(invoked, status=..., count=...) with the return value of tool_invoked. status is ok / failed / unknown. Use unknown when you do not know the result. The effect of the write is recorded as unknown (had_mutating_success can become null), and if the run does not succeed it is not retried automatically and is sent to a human for review. A write tool_invoked without a matching tool_effect also leaves the effect unknown (reads do not need one)
3. If you decided not to run an operation, call tool_denied(tool, reason=...). After even one call, the result of the whole run is denied even if the function returns normally (conditions earlier in the evaluation order, such as audit persistence failures, cancellation and deadlines, take precedence). How a denied run is handled afterwards follows the workflow settings

A run in which saving even one audit record failed ends as failed/audit_persist_failed. An audit refused after cancellation, the deadline or the end of the workflow run does not count as a save failure; the result follows the cancellation or deadline (up to v3.0.12, an audit refused before the runner knew about the stop ended the run as failed/audit_persist_failed).

Only strings, integers, finite floats and booleans can be passed in the audit metadata (**extra). The SDK drops anything else (NaN / Infinity, lists, dicts, None) and logs the names of the dropped keys as a warning (v3.0.13+; values are not logged).

Recording audits does not decide by itself whether an operation is allowed. Your agent implements and follows the governance of its own operations (importing the SDK's audit API or the Policy Module does not stop anything). Never start an external operation without being able to check whether it is allowed (see When there is nothing to do or you cannot check).

effect_kind is read / list / query / fetch for reads and write / create / update / delete / send / mutate for writes. Only strings, numbers and booleans are kept in the audit metadata. Do not put bodies, tokens or argument values in it.

Sending externally (idempotent write)Python

async def run(context):
    await context.ensure_can_act()  # do not start new external operations after cancellation or the deadline
    action_ref = f"send:{context.operation_key[:16]}"  # idempotency key that stays the same across resends of the step
    body = (context.work_dir / "reply.md").read_text(encoding="utf-8")

    invoked = await context.audit.tool_invoked("send_mail", effect_kind="send", action_ref=action_ref, target="mail")
    try:
        receipt = await mail_client.send(idempotency_key=action_ref, body=body)  # your sending logic
    except Exception as e:
        # never treat a failure with an unknown result as "not sent"
        await context.audit.tool_effect(invoked, status="unknown", action_ref=action_ref, error=type(e).__name__)
        raise
    await context.audit.tool_effect(invoked, status="ok", count=1, action_ref=action_ref)
    # declare receipt.md in the contract's outputs (undeclared files are not collected)
    (context.output_dir / "receipt.md").write_text(f"# receipt\n\n{receipt}\n", encoding="utf-8")

An agent that writes externally like this sets the contract's effect_mode to idempotent_write or similar. To prepare for resends, use operation_key and a stable action_ref as the idempotency key of the external system. Do not call a write again just because of a network error when its result cannot be checked.

When there is nothing to do or you cannot check

skip() and policy_unavailable() only record the decision; they do not stop your agent. Return from the function after calling them.

Python

async def run(context):
    targets = load_targets(context.work_dir)
    if not targets:
        await context.skip("no_target")   # nothing to process
        return

    if not await can_check_permissions():
        await context.policy_unavailable()  # could not check whether the operation is allowed before acting
        return
    ...

• skip(reason) (no_target / not_applicable / nothing_to_do): the step ends as skipped and the workflow proceeds to the next step
• policy_unavailable(): the step ends as skipped/policy_unavailable and does not proceed. The platform checks again within its automatic retries, and sends it to a human for review when there are no automatic retries or the limit is reached. If the run has called a write tool_invoked even once, the result is failed/skip_after_effect

Cancellation and deadlines

• Call await context.ensure_can_act() before starting an external operation. After cancellation or the deadline it raises WorkflowError
• The runner takes the margin for stopping, collecting and saving from the collection deadline, and stops your agent at the earlier of the execution deadline and "collection deadline − margin". context.deadline_remaining() tells you the time left
• To stop your agent, the runner sends SIGTERM to the child process group and, if processes remain after the grace periods (10 seconds each), SIGKILL. Once it confirms they have stopped, it collects the outputs written by then. The result follows the reason for the cancellation, deadline or stop (see Execution result)
• Once the runner knows about a cancellation, deadline or pod stop, is_cancel_requested() returns True and ensure_can_act() / connection_token() raise WorkflowError (from v3.0.6 the stop is also delivered in replies to other calls such as audits)
• After a cancellation the platform also refuses saving outputs and the result. The runner ends as a stop and exits with 0 without saving the result; the platform closes the run as cancelled (v3.0.13+)

Connection tokens

The access tokens of the connections placed in the contract's connection_slots are passed at startup in the environment variable ASTER_CONNECTION_<SLOT>_TOKEN (the slot name in upper case, with - replaced by _; therefore slots a-b and a_b cannot be used together). These tokens can expire after about an hour, so in long runs use connection_token() before calling external APIs (v3.0.5+).

Python

token = await context.connection_token("gitlab")      # {access_token, token_type, expires_at}
headers = {"Authorization": f"Bearer {token['access_token']}"}

The connection endpoint is not passed together with the token. The platform database connection is not passed to steps either, so pass the connection URL in a fixed environment variable.

It returns the cached value until less than min_ttl seconds (default 300) remain before expiry, then fetches a new one from the platform. Tokens with an unknown expiry are refreshed every 600 seconds. A slot not bound at startup, a revoked connection or one awaiting re-authorization, and any call after cancellation or the deadline raise WorkflowError. Never put tokens in logs or exception messages.

Runs that follow a review decision

When a step runs again because a reviewer sent the work back to it, or the next step runs after a reviewer advanced the work, context.gate_decision holds the decision on that review request (v3.0.9+). It is None for every other run (the first step when a workflow starts, and steps started without a review).

Key	Description
gate_id	ID of the review request
decision	send_back (sent back) or proceed (advanced)
decided_by	human (a reviewer decided) or system (decided automatically when the condition of an external-condition wait was met; in the Marketplace edition it is never decided by a deadline)
note	The reason for sending back, or the comment added when advancing, as written by the reviewer (leading and trailing whitespace removed). None when nothing or only whitespace was written, or the decision was automatic
decided_at	When the decision was made (ISO 8601; None if unknown)

Add the reason for a send-back to your agent's instructions so that it produces output that fixes what was asked. Up to v3.0.8 the reason never reached the agent, so the agent could not act on it.

Python

async def run(context):
    instructions = (context.work_dir / "instructions.md").read_text(encoding="utf-8")

    review = context.gate_decision
    if review and review["decision"] == "send_back" and review["note"]:
        instructions += f"\n\n## Requested changes\n\n{review['note']}\n"
    ...

• The start record (context.manifest) includes only gate_id / decision / decided_by; the text of the reason or comment is not included
• On platform versions that do not return gate_decision, it is always None and nothing else changes. To use it, rebuild your image with this SDK version and have it certified again

Execution result (outcome)

The result is recorded as a value that describes how the run ended. Success means your agent ran to completion; failed writes or unobserved effects do not lower it. First, the table below is evaluated from the top:

outcome / reason code	When
failed / audit_persist_failed	An audit record could not be saved
stopped / cancelled	The run was cancelled
timeout / execution_timeout	The deadline passed
stopped / terminal_elsewhere	The run had already been closed elsewhere
failed / runner_stopped	The pod was stopped
denied / <reason_code>	A denial was recorded with tool_denied (the reason_code of the first denial; default tool_denied)
failed / agent_error	Your agent function raised an exception, or the entry point could not be imported
failed / output_invalid	A declared output is invalid or unreadable
failed / skip_after_effect	policy_unavailable() was called in a run that called a write tool_invoked
skipped / <reason>	skip() or policy_unavailable() was called (when no successful write was observed)
failed / required_output_missing	A required output is missing
completed / completed	The function returned normally

After that, the runner's own state can replace the result.

outcome / reason code	When
failed / runner_restarted	The runner restarted and could not observe the result of your agent
failed / artifact_save_failed	Saving the outputs partly failed (replaces completed / skipped)
failed / cleanup_uncertain	The previous agent process could not be stopped completely
failed / input_integrity	An input file digest, size or path did not match (your agent is not started)

If a cancellation or the deadline arrives before your agent starts (including while the inputs are being staged), the run ends as stopped/cancelled / timeout/execution_timeout without starting your agent. incomplete is part of the shared vocabulary, but the workflow runner does not produce it.

What was written is kept in the result's counts / tools and had_mutating_success (true = a successful write was observed / false = confirmed no effect / null = unknown). Unsuccessful runs with null are not retried automatically and are sent to a human for review.

Debugging

• Your agent's standard output and standard error go to the pod log. Messages from context.log() go to the runner log (the pod's standard error). From v3.0.13, run() sends SDK logs at INFO and above to standard error when the process has not configured logging (up to v3.0.12, context.log() did not appear with the recommended start command). If you configure logging yourself, your configuration applies
• By default, only the class name of an exception raised by your agent is written to standard error and to the result (error_class). To print the full traceback, set ENV ASTER_WORKFLOW_DEBUG_TRACEBACK=1 in the image's Dockerfile (bodies or tokens may appear in the log, so remove it when you are done investigating)
• Two samples (drafting a reply, a mock send) are included under docs/examples/ in the source distribution (sdist) on PyPI

Next steps

SDK API Reference — Workflow Module

Complete specification of run / WorkflowContext / exceptions / results

ガイドを見る

Deploy Guide

Building the image and registering it in the admin console

ガイドを見る