Skip to main content

Patterns

Two ways to combine the routes on Run a workflow from your system with what a workflow author builds in Studio: a workflow that reads files your system dropped in Amazon S3, and one that ends by training the next version of a knowledge graph. api is the shell function from step 1 of that page and call the helper from its Python example.

Files dropped in S3 are picked up by a workflow​

Your system writes files to a prefix in a bucket you own; when a run starts, the workflow reads that prefix through the AWS connector's role in your account, copies the objects into the run, and the steps work on them. Nothing watches the bucket: a run reads the prefix when it starts, and a schedule or your POST starts it.

Before you start: a workflow author has built an AWS connector with a recorded passing Test connection, assigned it to the agent that owns the reading step (Agents: configure and activate), and activated a workflow with an s3_prefix parameter naming that connector (Workflows: build). Your key needs workflows:read and runs:create for it.

  1. Write the files under a prefix of your choosing, for example s3://<your-bucket>/incoming/2026-09-27/. Any object type is fine. Finish writing before you start the run: the copy happens once, at launch, and objects added afterwards are not seen by that run.

  2. Read the parameter contract.

    api "https://$STUDIO/api/v1/workflows/<workflow_id>/parameters"
    {"name": "drop", "type": "s3_prefix", "required": true, "description": "The batch to process",
     "connector": {"connector_id": "<connector_id>", "name": "Intake bucket", "region": "eu-west-1"},
     "bucket": "<your-bucket>", "prefix_default": "incoming/",
     "constraints": {"max_objects": 500, "max_bytes": 2147483648}}

    connector is the AWS connector whose role reads the bucket, with its default region. With a fixed bucket you pass only a prefix; without one you pass a full s3://bucket/prefix, which the role must be allowed to read. prefix_default applies when you omit the parameter. constraints are the most a run copies in: 500 objects and 2 GiB by default, up to 5,000 and 20 GiB.

  3. Start the run by API, with the prefix as the value:

    api -X POST "https://$STUDIO/api/v1/workflows/<workflow_id>/runs" \
      -H "Idempotency-Key: intake-2026-09-27-batch-1" \
      -H "Content-Type: application/json" \
      -d '{"parameters": {"drop": "incoming/2026-09-27/"}, "client_reference": "batch-2026-09-27-1"}'
    The parameter hasYou passExample
    A fixed bucketA prefix inside it; the s3://<that bucket>/… form is accepted and strippedincoming/2026-09-27/
    No fixed bucketA full s3://bucket/prefixs3://<your-bucket>/incoming/2026-09-27/

    A trailing / is added if missing. The value must not start with /, contain . or .. segments or control characters, or exceed 900 characters after the bucket. A prefix in another bucket ("must be a prefix inside bucket …") or a bare prefix where a URI is needed ("must be a full s3://bucket/prefix because the parameter has no fixed bucket") answers 400 invalid_parameters with the parameter's name and reason.

    Or by schedule from Studio: a cron(...) or rate(...) expression in UTC with fixed parameter values suits a stable prefix (incoming/) your system keeps filling and the workflow drains, not a per-batch prefix. Scheduled runs belong to the person who created the schedule and are not visible to your key on GET /runs; trigger by API when your system must follow the run (Workflows: run, schedule, approve).

  4. At launch the API checks readiness (the connector exists, is not flagged as needing attention and has a recorded passing Test connection; otherwise 409 workflow_not_ready with details.reasons), then copies every object under the prefix into the run's inputs/drop/ folder (named after the parameter), up to the caps. inputs_manifest lists what arrived; a step's objective can refer to {{params.drop}}. A failed copy ends the run before any step (Notes).

  5. Collect the outputs from the run's outputs/ folder in the deployment's own workspace storage, through GET /runs/{run_id}/outputs, /outputs/{output_id}/content and /outputs.zip; a final step that writes outputs/result.json appears parsed as result. The platform writes nothing back to your bucket. If you want results there, the author has the task's agent write them through the same connector (the AWS SDK in its sandbox, with the role's temporary credentials), and the role's policy must allow s3:PutObject on the destination.

Train a knowledge graph from a workflow run​

A workflow can end by training the next version of a knowledge graph (AMPG) from what its steps learned. Started through the API, the run builds a training document, a person may review it in Studio's Inbox, and a new version is created and, if the author chose so, adopted by the agents that use the graph.

Before you start: the graph has a Ready version with training documents, and the author has configured the node (the target graph, the source steps, Rewrite or Append, Hold the new PDF for my review before creating the version, on by default, and Update all agents using this AMPG, off by default); see Workflows: knowledge-graph update node. Your key needs runs:create and runs:read.

  1. Start the run as on any workflow; the training happens in its last steps.

    api -X POST "https://$STUDIO/api/v1/workflows/<workflow_id>/runs" \
      -H "Idempotency-Key: kb-refresh-2026-09-27" \
      -H "Content-Type: application/json" \
      -d '{"parameters": {"drop": "incoming/2026-09-27/"}, "client_reference": "kb-refresh-2026-09-27"}'

  2. Watch the two steps the node becomes on GET /runs/{run_id}/steps: the build step, with the node's slug and node_type ampg_update, reads the sources and the base version's documents and writes the new training document; the publish step, <slug>-publish with node_type ampg_publish and derived: true (the runtime adds it; it is not a node of the version), depends on it and creates the version. Each has its own status, timings and result_summary.

  3. If the document is held for review, the build step finishes and parks: GET /runs/{run_id} reads waiting_for_approval with counts.awaiting_approval 1, /approvals lists the step under pending with "phase": "result", and /steps shows "awaiting": {"phase": "result", "since": "…"}. A person opens the Inbox card, reads the PDF and chooses Approve, Send back or Stop; the API cannot. Once approved, approved_by and approved_at name that person, the publish step starts and the run returns to running. Keep polling every 5 seconds; a long wait here is a person, not a fault.

  4. Rewrite or Append. The publish step creates version N+1 from base N: Rewrite (the default) rewrites version N's documents and the learnings from the connected steps into one new training document, the whole next version; Append keeps version N's documents and adds one document with only the learnings. The base is the graph's highest Ready version when the run publishes, not when the workflow was saved: the builder's "Starts from v<N> · <k> documents" line is resolved again at run time.

  5. Adoption. With Update all agents using this AMPG on, every agent that uses the graph moves to the new version once it is Ready, and the event below lists them in adopted_agents. With it off, the version is Ready, agents keep the version they pin, and Studio posts an Inbox card from which a person publishes and adopts it (Knowledge graphs: use, move agents, delete).

  6. Read the new version. The API has no knowledge-graph routes; the run tells you in three places: the publish step's result_summary once it is completed; the event ampg_version_created on GET /runs/{run_id}/events, whose payload carries kb_id, version, key (the training document's workspace key), adopted_agents, job_ids, base_version, base_documents and mode; and the graph's versions table in Studio, where the row appears as Building and then Ready (Knowledge graphs). The approval started a new segment, so read segments and poll the latest from after_seq=0.

    api "https://$STUDIO/api/v1/runs/<run_id>/events?after_seq=0&limit=500"
    events = call("GET", f"/runs/{run['run_id']}/events", params={"after_seq": 0, "limit": 500}).json()
    created = [e for e in events["events"] if e["type"] == "ampg_version_created"]
    if created:
        p = created[-1]["payload"]
        print("graph", p["kb_id"], "version", p["version"], "from", p["base_version"], "mode", p["mode"],
              "adopted", p["adopted_agents"])

What you should see​

PatternWhat the run reports
Files in S3202 with the run running; parameters.drop normalised with its trailing /; inputs_manifest listing each copied object with name and size; on /steps the reading step running, then completed with a result_summary; the files on /outputs once completed
Knowledge graphrunning while the upstream steps and the build work; waiting_for_approval while the document is held; running again after the approval; completed once the version is Ready (and, with the toggle on, the agents moved). In Studio's Inbox the review card, then the publish card, appear for everyone assigned to the deployment

Notes​

  • The connector's session is 3,600 seconds by default, between 900 and 43,200 seconds, capped by the role's own maximum session duration.
  • 409 workflow_not_ready with "connector … needs attention": Studio has flagged the connector and the reason follows the colon; the author fixes it and runs Test connection (Connector versions, Test connection and guides).
  • input_materialisation_failed (the run ends failed with this error.code before any step runs: role denied, empty prefix or a cap exceeded): check the role's permission policy against the bucket, that the objects were written before the run started, and the object count and size against constraints. A run that processed an older batch means the schedule passed a fixed prefix: trigger by API with a per-batch prefix, or point the schedule at a prefix your system drains.
  • A graph holds 50 versions and builds one at a time: a publish step that finds another version building fails; wait, then start a new run. A Rewrite with no documents behind the base fails rather than rewrite nothing (the builder warns "No training documents found for v<N>"); the author builds a version from documents first, or switches the node to Append.
  • After 3,600 seconds the publish step fails with no ampg_version_created event while the version keeps building; check the graph's versions table in Studio, and if it reads Failed there, open it for the reason.