Workflows via de API¶

Het concept achter workflows staat in Workflows.

Op deze pagina leg je een bedrijfsproces vast als workflow, controleer je het, start je een run en volg je die tot het einde. Een workflow is een graaf die je opslaat: de knooppunten zijn stukken werk -- een agent, een losse prompt, een aanroep van een tool, een voorwaarde, een samenkomst -- en de verbindingen geven de uitvoer van het ene knooppunt door aan het volgende. Core voert de graaf uit die je hebt opgeslagen, stap voor stap op de achtergrond, dus een proces met een agent, een opzoeking en een tweede agent naast elkaar loopt zonder dat jouw applicatie ervoor hoeft te blijven draaien.

Je sleutel heeft de capability workflows nodig, en die heeft runs en threads nodig: elke agent- of promptstap is een achtergrondrun op een eigen gesprek. Een toolstap heeft ook tooling nodig. Je operator zet workflows aan voor de hele installatie; zolang dat niet is gebeurd, antwoordt elke route hieronder 404. Hetzelfde doe je met de muis in Workflows in de console.

Een workflow schrijven¶

curl -s -X POST "$CORE_URL/workflows" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"name": "intake",
         "description": "Een bericht indelen en een antwoord opstellen",
         "input_schema": {"type": "object",
                          "properties": {"message": {"type": "string"}},
                          "required": ["message"]},
         "graph": {
           "nodes": [
             {"id": "start", "kind": "start"},
             {"id": "classify", "kind": "agent", "config": {"agent": "triage"},
              "input": "run.input.message"},
             {"id": "route", "kind": "condition"},
             {"id": "draft", "kind": "agent", "config": {"agent": "writer"},
              "input": "run.input.message"},
             {"id": "reply", "kind": "end"},
             {"id": "spam", "kind": "end", "input": "'\''dropped'\''"}
           ],
           "edges": [
             {"source": "start", "target": "classify"},
             {"source": "classify", "target": "route"},
             {"source": "route", "target": "draft",
              "when": "nodes.classify.output != '\''spam'\''"},
             {"source": "route", "target": "spam"},
             {"source": "draft", "target": "reply"}
           ]}}'

Een geslaagd verzoek antwoordt 201 en een Location, met de workflow en zijn eerste versie. Een naam is uniek binnen je organisatie: een tweede workflow met de naam intake geeft 409.

Een run die een workflow- of map-knooppunt startte, is een gewone run met "trigger_kind": "workflow"; GET /workflow_runs?parent_run_id=... toont de runs die één run startte. Elk daarvan is een eigen start, één gebruik van workflow_run, dus een map over honderd items kost honderd starts naast de run waarin hij staat. Een map neemt geen input en geen retry. Hoe zulke runs er in de console uitzien en welke grenzen gelden, staat in Workflows in de console.

Elk knooppunt kan een input opgeven, een JMESPath-expressie over de context van de run: run.input en nodes.<id>.output voor elke stap die klaar is (nodes.<id>.error.code voor een stap die mislukte). Zonder input krijgt een knooppunt de uitvoer van zijn ene voorganger, en een join een object met per voltooide tak een sleutel. Een tool-, agent- of prompt-knooppunt mag timeout_seconds en retry zetten ({"max_attempts": 3, "backoff_seconds": 10}). Een tool-knooppunt mag retry alleen zetten als de registratie van de tool zegt dat twee keer versturen veilig is (retry_safe); anders weigert het opslaan de retry van het knooppunt. Een verbinding met "on_error": true is het vangnet van een knooppunt: een mislukte stap volgt die in plaats van zijn andere verbindingen.

Alles wordt gecontroleerd bij het opslaan. Een ontbrekende agent, een tool die je sleutel niet mag aanroepen, een kringloop, twee verbindingen die buiten een join samenkomen of een expressie die niet te lezen is, geeft 422 met het veld erbij, zoals graph.nodes[1].config.agent -- niet drie dagen later een mislukte run. Een sleutel zonder de capability tooling, of met een beleid dat een agent vereist, mag helemaal geen tool-knooppunt hebben: een toolstap loopt buiten elke agent, dus het opslaan geeft 422 op graph.nodes[i].config.tool. Een toolstap kan ook geen agent noemen (agent__<naam>): draai een agent met een stap van soort agent. Verwijzingen zijn namen: een waarde in de vorm van een id, zoals agt_01J..., wordt geweigerd. Om te controleren zonder op te slaan, neemt POST /workflows/validate dezelfde graph en input_schema.

Een run starten¶

curl -s -X POST "$CORE_URL/workflows/wf_01J.../runs" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -H "Idempotency-Key: ticket-4711" \
    -d '{"input": {"message": "Mijn factuur klopt niet."}}'

Het antwoord is 202 met de run, queued. Een sleutel start een run als zichzelf: elke stap handelt als jouw sleutel, wordt beslist door het beleid van jouw sleutel op het moment dat hij loopt, en besteedt het budget van jouw sleutel. Het starten van een run is één gebruik van de capability op je rekening; wat de stappen laten lopen wordt gerekend zoals altijd. version legt een oudere versie vast, subject_id noemt de persoon voor wie elke stap is, labels labelt elk gesprek dat de stappen openen en elke run die ze starten (overerving), en dezelfde Idempotency-Key antwoordt met de run die hij startte in plaats van een nieuwe te starten; dezelfde waarde gebruikt door een andere API-sleutel geeft 409. Een Idempotency-Key die begint met schedule:, trigger:, webhook: of task: geeft 422: die vier voorvoegsels zijn de sleutels die Core geeft aan de runs die een planning, een trigger of een taak start. Invoer die groter is dan de installatie toestaat, geeft 413. Heeft je organisatie al zoveel open runs als de installatie toestaat, of heeft je sleutel zijn maandplafond bereikt, dan is het antwoord 429.

Starten als er iets gebeurt¶

Een trigger start een run telkens als Core een gebeurtenis van één soort voortbrengt, dezelfde gebeurtenissen die de feed en webhooks dragen. Noem de workflow, de soort, en eventueel welke gebeurtenissen van die soort en wat de invoer van de run is:

curl -s -X POST "$CORE_URL/workflow_triggers" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"workflow": "intake",
         "event_kind": "knowledge.source.ingested",
         "filter": {"status": "completed"},
         "input": "{source: event.refs.source_id}"}'

filter is gelijkheid over de refs van de gebeurtenis: elke naam die je geeft moet er staan met precies die waarde, en {} past op elke gebeurtenis van de soort. input is een JMESPath-expressie over event (id, type, refs, subject_id, created_at); laat je hem weg, dan is de invoer van de run dat gebeurtenisdocument. Elke run handelt als de sleutel die de trigger schreef en heeft "trigger_kind": "event"; GET /workflow_runs?trigger_id=wft_... geeft de runs die één trigger startte. Eén gebeurtenis start per trigger hoogstens één run, hoe vaak hij ook wordt afgeleverd.

Voor wie een run is, volgt de gebeurtenis. Een trigger die niemand noemt start elke run voor de persoon bij wie de gebeurtenis hoort, dus een run uit een gebeurtenis van Anna is van Anna: alleen haar subjecttoken en je eigen sleutels lezen hem, en hij wordt met haar gewist. Een gebeurtenis die bij niemand hoort start een run voor niemand. Een trigger die in subject_id een persoon noemt start runs voor haar uit haar eigen gebeurtenissen en uit gedeelde, en een gebeurtenis die bij iemand anders hoort past er helemaal niet op.

Nieuwe mail¶

Een mailkoppeling waarvan de versie een watch instelt (folder, poll_seconds van minstens 60, tenzij je beheerder een andere ondergrens instelde, en enabled, geschreven met POST /mail_connectors of PATCH /mail_connectors/{id}) brengt voor elk nieuw bericht in die map één gebeurtenis mail.received voort. De gebeurtenis draagt het id van de koppeling in mail_connector, de folder en de uid van het bericht, nooit een adres, een onderwerp of een tekst. Ze hoort bij niemand, dus de run die ze start is voor niemand, tenzij de trigger een persoon noemt. Filter op de koppeling om voor elk nieuw bericht een workflow te starten:

curl -s -X POST "$CORE_URL/workflow_triggers" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"workflow": "triage-mail",
         "event_kind": "mail.received",
         "filter": {"mail_connector": "mail_01J..."},
         "input": "{folder: event.refs.folder, uid: event.refs.uid}"}'

De agentstap van de workflow leest het bericht zelf met de tool mail__<naam>__read van de koppeling. Mail die al in de map stond toen de watch begon, start geen run. Hoe je de koppeling en haar watch instelt, staat in Mailkoppelingen via de API.

Grenzen en mislukte starts¶

Twee grenzen voorkomen dat een trigger zichzelf voedt. De soorten die elke agentstap voortbrengt (response.*, run.*, thread.*) worden als event_kind geweigerd. En een trigger op een workflowgebeurtenis, bijvoorbeeld workflow.run.completed, mag runs maar een paar diep aan elkaar rijgen (drie, tenzij je beheerder anders instelde): de run daarna wordt niet gestart.

Niets houdt een workflow tegen die schrijft naar een datastore waar zijn eigen trigger op luistert: elke run voedt de volgende tot de limiet aan starts per uur (rate_limited, 600 tenzij je beheerder anders instelde) hem stopt.

Een start die niet kon doorgaan wordt niet opnieuw geprobeerd. Hij staat in je audittrail als workflow.trigger.failed met een korte code:

Een organisatie heeft hoogstens 50 triggers, gebeurtenis- en webhooktriggers samen, tenzij je beheerder anders instelde; een nieuwe daarboven krijgt 429. Met PATCH /workflow_triggers/{id} en {"enabled": false} zet je een trigger uit, en DELETE verwijdert hem; de runs die hij startte blijven. Elke sleutel die een trigger kan zien mag hem uitzetten, ook nadat de sleutel van de trigger zelf is ingetrokken. Weer aanzetten met {"enabled": true} is voor de sleutel waaronder de trigger runs start: een andere sleutel krijgt 403 met not_the_acting_key.

Starten vanuit een ander systeem¶

Een webhooktrigger geeft een workflow een adres. Een systeem van jou stuurt er JSON naartoe, ondertekend met een geheim, en elk verzoek start een run met die JSON als invoer. De afzender heeft geen API-sleutel nodig.

curl -s -X POST "$CORE_URL/workflow_triggers" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"workflow": "intake", "source": "webhook",
         "input": "{message: webhook.body.text}"}'

Het antwoord bevat de url van de trigger en, alleen deze keer, zijn secret (whsec_...). Bewaar het geheim nu: geen latere leesactie toont het, en POST /workflow_triggers/{id}/rotate-secret antwoordt met een nieuw geheim, waarna alleen het nieuwe werkt. De url is $CORE_URL/workflow_webhooks/<organisatie-id>/<trigger-id> en blijft dezelfde zolang de trigger bestaat. Een webhooktrigger krijgt geen event_kind en geen filter. input is een JMESPath-expressie over webhook (body, de JSON die je stuurde; delivery; received_at); laat je hem weg, dan is de invoer van de run de body zelf. Elke run handelt als de sleutel die de trigger schreef, is voor de persoon in subject_id van de trigger of voor niemand, en heeft "trigger_kind": "webhook".

De afzender ondertekent elk verzoek zoals Core de webhooks ondertekent die het jou stuurt: een header Artific-Signature met t=<unix-seconden>,v1=<handtekening>, waarbij de handtekening de hexadecimale HMAC-SHA256 is, onder het geheim, van het tijdstip, een punt en de exacte bytes van de body.

import hashlib, hmac, json, time
import httpx

def post_to_core(url: str, secret: str, payload: dict, delivery: str) -> int:
    body = json.dumps(payload).encode()
    stamp = int(time.time())
    digest = hmac.new(secret.encode(), f"{stamp}.".encode() + body, hashlib.sha256).hexdigest()
    answer = httpx.post(
        url,
        content=body,
        headers={
            "Artific-Signature": f"t={stamp},v1={digest}",
            "Idempotency-Key": delivery,
            "Content-Type": "application/json",
        },
    )
    return answer.status_code

Stuur precies de bytes die je ondertekende. Idempotency-Key is je eigen naam voor de levering, hoogstens 128 tekens zonder spaties: probeer je het na een time-out opnieuw, onderteken dan opnieuw en stuur dezelfde naam, en elke poging is één run. Hetzelfde ondertekende verzoek dat twee keer aankomt start ook één run.

Een 202 zegt dat het verzoek is aangenomen, niet dat de run is gestart. Of hij is gestart staat in je audittrail, zoals bij een trigger op een gebeurtenis: workflow.trigger.fired met de run, of workflow.trigger.failed met een code. De code expired_unstarted betekent dat Core het verzoek aannam en de run niet binnen een uur kon starten; hij wordt niet alsnog gestart, dus stuur het opnieuw. Webhooktriggers hebben een eigen aandeel starts per uur, even groot als dat van de triggers op gebeurtenissen (600, tenzij je beheerder anders instelde), en GET /workflow_runs?trigger_id=wft_... geeft de runs die één trigger startte. Alleen de sleutel waaronder de trigger runs start mag het geheim vervangen of de trigger weer aanzetten; elke sleutel die hem kan zien mag hem uitzetten. Een systeem dat alleen op zijn eigen manier kan ondertekenen, zoals een repositoryhost of een betaalprovider, kan niet rechtstreeks naar het adres sturen: laat een kleine dienst van jou zijn aanroep ontvangen, controleren en ondertekend zoals hierboven naar Core sturen.

Een agent laten starten¶

Een agent kan midden in een gesprek een workflow starten en hoort het wanneer die klaar is. Noem de workflow in de tools van de agent, of in de tools van één verzoek:

curl -s -X POST "$CORE_URL/responses" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"model": "agent:support", "thread_id": "thr_01J...",
         "input": "Neem deze klant aan: Anna, anna@example.com",
         "tools": [{"type": "workflow", "name": "intake"}]}'

Het model krijgt een tool aangeboden die workflow__intake heet. De argumenten zijn de invoer die de nieuwste versie van je workflow aanneemt, dus het input_schema van die versie moet een object zijn ("type": "object"). De beschrijving is die van de workflow zelf: schrijf daar wanneer de agent hem moet gebruiken.

Een workflow draait altijd op de achtergrond. De aanroep antwoordt meteen met een taak-id en het gesprek gaat door. De run start op de nieuwste versie van de workflow, als de sleutel en voor de persoon van de beurt, en heeft "trigger_kind": "task". Als de run eindigt, komt de uitvoer als resultaat van de taak in de thread, en Core start een beurt om erop te antwoorden als er op dat moment niemand iets vraagt. Dat kan dagen later zijn: een run die een agent startte, duurt zo lang als elke run van de workflow mag duren, ook als een beslissing op een manager wacht. Intussen toont GET /tasks/{id} de taak met zijn workflow_run_id en, als progress, de stappen waarop de run wacht.

Je sleutel heeft naast workflows ook tasks nodig, en de beurt moet op een thread zijn opgeslagen: een resultaat moet ergens terecht kunnen. Een beurt die dat niet is, of een sleutel zonder tasks, draait zonder de tool en wordt niet geweigerd. Een sleutel zonder workflows krijgt een 400 die de regel noemt, en een workflow die niet bestaat een 404. Passen de argumenten die het model schreef niet bij het invoerschema, dan start er niets en leest het model dat ze niet passen, zodat het het opnieuw kan proberen.

De thread verwijderen stopt de run niet. Een run kan al iemand iets gevraagd hebben of een ander systeem hebben aangeroepen, dus hij loopt door tot zijn einde en alleen het resultaat vervalt; annuleer de run zelf als je hem wilt stoppen. Een thread heeft hooguit acht open taken (tenzij je beheerder of je sleutel minder zegt), en een workflow die een week op een antwoord wacht, telt die hele week mee.

Volgen¶

curl -s "$CORE_URL/workflow_runs/wfr_01J..." -H "Authorization: Bearer $CORE_KEY"

Een run is queued, running, waiting zolang er niets open is behalve een wait of een decision, en daarna completed met zijn output, failed met een error_code, cancelled of expired. Een run die op zijn deadline_at nog open is, verloopt: standaard 30 dagen nadat hij startte, eerder als je organisatie inhoud korter bewaart. Zijn steps zeggen wat elk knooppunt ontving en opleverde, hoe vaak het geprobeerd is en, bij een agentstap, de response_id en thread_id waarop het liep, die je opent als elke achtergrondrun. GET /workflow_runs?workflow_id=...&status=running toont de runs, nieuwste eerst. Een planning start runs op een klok, en ?trigger_id= beperkt de lijst tot de runs die één planning startte. De gebeurtenissenstroom en webhooks dragen workflow.run.started, .completed, .failed, .cancelled en .expired, en workflow.decision.requested, .decided en .expired voor elke beslissing.

POST /workflow_runs/{id}/cancel stopt een run en elke open stap, de agentruns die hij startte inbegrepen; nogmaals vragen antwoordt met de geannuleerde run, en een run die al op een andere manier eindigde geeft 409.

Iemand laten beslissen¶

Een decision-knooppunt houdt de run tegen tot er één antwoord komt. Het noemt de uitkomsten waaruit gekozen wordt, elk met precies één uitgaande verbinding met "outcome": "<naam>", en wie mag antwoorden:

{"id": "approve", "kind": "decision",
 "config": {
   "outcomes": ["approve", "reject"],
   "instruction": "join(' ', ['Terugbetaling voor', run.input.customer])",
   "form": {"type": "object", "properties": {"note": {"type": "string"}}},
   "assignee": {"kind": "members"},
   "deadline_seconds": 86400,
   "on_deadline": {"outcome": "reject"}}}

instruction is een expressie die de tekst geeft die de toegewezene leest; naast run en nodes leest ze de eigen invoer van het knooppunt als input. Ze wordt één keer uitgewerkt, op het moment dat de stap gaat wachten.

curl -s "$CORE_URL/workflow_decisions" -H "Authorization: Bearer $CORE_KEY"

curl -s -X POST "$CORE_URL/workflow_runs/wfr_01J.../steps/wfs_01J.../decision" \
    -H "Authorization: Bearer $CORE_KEY" -H "Content-Type: application/json" \
    -d '{"outcome": "approve", "fields": {"note": "Factuur gecontroleerd."}}'

GET /workflow_decisions toont de wachtende beslissingen die de aanroeper mag beantwoorden, nieuwste eerst, elk met zijn id (de stap), run_id, workflow, instruction, outcomes, form en deadline_at. Het antwoord noemt een van de uitkomsten en, als het knooppunt een form heeft, fields die erin passen; al het andere geeft 422 met het veld erbij. Alleen de toegewezene mag antwoorden: een andere aanroeper die de run kan lezen krijgt 403, en wie dat niet kan 404. Een beslissing wordt één keer beantwoord: een tweede antwoord, en een antwoord na de deadline, geeft 409. De stap is dan klaar met {outcome, fields, decided_by} en de run volgt de verbinding van die uitkomst.

Als deadline_seconds voorbijgaat zonder antwoord, beslist on_deadline: {"outcome": "..."} neemt die uitkomst, "error" laat de stap mislukken met decision_expired voor zijn on_error-verbinding, en "fail", de standaard, laat de run mislukken met die code. Een wait, en een decision die een persoon of een sleutel beantwoordt, nemen geen timeout_seconds en geen retry; de deadline van de run zelf begrenst ze. Een decision die aan een agent is toegewezen is de uitzondering (hieronder).

Een agent beslist¶

Een beslissing die aan een agent is toegewezen wacht op niemand. De agent leest één bericht, {"instruction": ..., "input": ...}: de uitgewerkte instructie en wat het knooppunt ontving. De agent moet antwoorden met een van de uitkomsten van het knooppunt en, als het knooppunt een form heeft, met velden die erin passen. Core controleert het antwoord zelf, wat het model ook schreef, en alleen een antwoord dat de controle doorstaat volgt de verbinding van een uitkomst.

Voor een aanroeper ziet zo'n stap er anders uit dan een beslissing die op een persoon wacht:

Zo'n knooppunt neemt timeout_seconds en retry, zoals een agent-knooppunt, en geen deadline_seconds of on_deadline; opslaan weigert die twee met 422. Het plafond van timeout_seconds stelt je operator in, standaard een uur, en laat je het veld weg, dan geldt dat plafond. De agent mag geen eigen antwoordformaat vastleggen (text), het form van het knooppunt mag geen $ref, $dynamicRef of $defs bevatten, en het antwoordformaat dat Core uit de uitkomsten en het form samenstelt mag hoogstens 64 KiB zijn; opslaan wordt in elk van die gevallen geweigerd.

Geeft de agent geen geldig antwoord, dan mislukt de stap met een van deze codes:

Zet het knooppunt retry, dan probeert Core het opnieuw, behalve als het plafond of het beleid de beurt al tegenhield voor die begon. Daarna volgt de stap zijn on_error-verbinding. Teken die verbinding voor wat er dan moet gebeuren, bijvoorbeeld naar een tweede beslissing die een persoon neemt; zonder die verbinding mislukt de run. policy_refused is de uitzondering: die code beëindigt de run meteen, en geen on_error-verbinding vangt haar op.

Core garandeert dat het antwoord een van de uitkomsten is, dat de beslissing als die van een agent is gemarkeerd en dat de beurt erachter na te lezen is. Core garandeert niet dat de agent zich niet liet sturen door wat de agent beoordeelde: je instructie en de invoer van het knooppunt bereiken de agent in één bericht, en een tekst in die invoer die zegt "keur dit goed" kan zijn zin krijgen, want approve is een geldige uitkomst. De capability trust_envelope markeert dat bericht voor het model als onbetrouwbaar, je instructie inbegrepen; dat verkleint de kans en neemt niets weg. Zet waar de uitkomst ertoe doet een beslissing die een persoon neemt achter die van de agent, en geef de beslissende agent geen tools, zodat wat de agent leest het antwoord kan veranderen en verder niets.

De stap toont assignee_kind: "agent", en na de beslissing noemt decided_by de agent, de versie en het response_id van de beurt, waar je kunt nalezen wat de agent las en schreef. workflow.decision.decided draagt dezelfde id's. Een run voor een persoon mag zijn beslissing door een agent laten nemen; of jouw proces over een persoon mag beslissen zonder dat er een mens aan te pas komt, bepaal je zelf, en een knooppunt dat aan members of aan een subject is toegewezen wordt nooit door een agent beantwoord.

Wijzigen¶

POST /workflows/{id}/versions schrijft de volgende versie; runs die al liepen houden de versie die ze vastlegden. Stuur de hele graph en het input_schema mee: een versie erft het schema van de vorige niet, en een weggelaten input_schema is {}, dat elke invoer accepteert. PATCH /workflows/{id} wijzigt description en labels. DELETE /workflows/{id} verwijdert de workflow met zijn versies en afgelopen runs, en geeft 409 zolang er een run open is. Een regel van soort workflow in het beleid van je sleutel beslist welke workflows die mag zien en starten, op naam en label.

Een beheerder kan een versie ook als concept opslaan, in de console of de API ervan. Geen sleutel, trigger of schema ziet of start een concept: een run zonder version start de gepubliceerde versie, en een run, een stap of een subworkflow die het nummer van het concept noemt, hoort dat de versie niet bestaat. Een workflow met alleen een concept bestaat niet voor je sleutel, een trigger of een schema tot hij gepubliceerd is. Een sleutel schrijft geen concept: een veld draft dat je sleutel meestuurt met POST /workflows of POST /workflows/{id}/versions wordt genegeerd, zoals elk ander veld dat die bodies niet kennen, en de versie wordt meteen gepubliceerd.

GET /workflows geeft de workflows die je sleutel mag zien. ?label=afdeling:salaris versmalt de lijst tot die met dat label; de parameter mag herhaald worden, herhalen betekent alle genoemde, en een naam herhalen vraagt om al die waarden. Een workflow waarvan het label meerdere waarden heeft, telt mee als de waarde daar een van is. Een waarde die geen naam:waarde is, antwoordt 422 met label erbij.

Het volledige contract staat in de API-referentie.