Metadata-Version: 2.5
Name: pegasus-workflows-sdk
Version: 0.36.2
Summary: Python SDK and CLI for authoring, packaging, and publishing Pegasus workflows.
Author: DolasDev
License: UNLICENSED
Keywords: automation,pegasus,temporal,workflows
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.11
Requires-Dist: httpx<1,>=0.27
Requires-Dist: mcp<2,>=1
Requires-Dist: temporalio<2,>=1.7
Requires-Dist: typer<1,>=0.12
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Provides-Extra: mcp
Description-Content-Type: text/markdown

# Pegasus Workflows SDK

`pegasus-workflows-sdk` is the Python SDK and CLI for authoring, packaging, and
publishing **Pegasus workflows** — Temporal workflows that automate
cross-domain operations (move lifecycle, billing follow-ups, dispatch
decisions) against the Pegasus public API.

Phase 1 ships the **developer flow**: write a workflow locally, run it against a
Dockerized Temporal, package it, and upload it. There is no server-side
execution yet — the API stores the artifact and lists it.

## Install

```
pip install pegasus-workflows-sdk
```

This installs the `pegasus-workflows` CLI. **Python 3.11+** is required. Pin the
version in your project's requirements for reproducible builds, e.g.
`pegasus-workflows-sdk==0.1.0`.

### Interim / unreleased install (git)

The repository is public, so you can install straight from a tagged commit
without waiting for a PyPI release — useful for an unreleased fix, or before the
first PyPI publish lands:

```
pip install "pegasus-workflows-sdk @ git+https://github.com/DolasDev/pegasus@sdk-python-v0.1.0#subdirectory=packages/workflows-sdk-python"
```

Swap the `@sdk-python-v0.1.0` tag for `@main` to track the latest unreleased
SDK. This clones the whole monorepo to build one subdirectory, so prefer the
PyPI install for everyday use.

## First-run setup

One command does the whole first-run bootstrap — seed a credential profile and
wire the authoring MCP server into your agent host:

```
pip install pegasus-workflows-sdk   # the MCP server ships in the base package
pegasus-workflows setup             # seeds ~/.pegasus/credentials (0600) + writes .mcp.json
```

`setup` is the front door the obvious `--setup` / `--configure` guesses point
at. It:

- seeds/updates a `~/.pegasus/credentials` profile at `0600` (delegates to
  `configure`; pick the profile with `--profile NAME`),
- writes the `pegasus` MCP-server stanza into `./.mcp.json` (Claude Code project
  config), never clobbering an existing `pegasus` entry without `--force`, and
- performs **no network calls** and writes the `api_key` **only** to the `0600`
  credentials file — never into `.mcp.json`.

Scriptable (zero prompts) when you pass everything as flags:

```
pegasus-workflows setup --profile qa --api-key vnd_... --api-root https://api.pegasus-qa.dolas.dev
pegasus-workflows setup --print-mcp-config      # emit the stanza to stdout, write nothing
pegasus-workflows setup --skip-mcp              # only seed the credential profile
```

Then start authoring (see [Quick start](#quick-start)). For other agent hosts or
manual wiring, see [Using the SDK with an AI coding agent](#using-the-sdk-with-an-ai-coding-agent).

## Quick start

```
pegasus-workflows init demo
cd demo
pegasus-workflows test demo
pegasus-workflows diagram                # prints a prompt — your coding agent draws workflow.mmd
pegasus-workflows package
pegasus-workflows push --profile default
```

> A workflow **diagram** (`<source_dir>/workflow.mmd`) is required to publish. `init`
> ships a starter one; `pegasus-workflows diagram` prints a prompt you feed to your
> own coding agent (Claude Code, Cursor, …) to draw it — no API key or extra needed.
> Business users view it in the Pegasus tenant UI to confirm the workflow matches
> their business rules. See [Visualizing workflows](#visualizing-workflows).

## Authoring

Import the Temporal authoring primitives from `pegasus_workflows` and mark your
workflow class with `@pegasus_workflow`:

```python
from datetime import timedelta
from pegasus_workflows import activity, pegasus_workflow, workflow

@activity.defn
async def greet(name: str) -> str:
    return f"Hello, {name}!"

@pegasus_workflow(name="demo", version="0.1.0")
class HelloWorkflow:
    @workflow.run
    async def run(self, name: str = "world") -> str:
        return await workflow.execute_activity(
            greet, name, start_to_close_timeout=timedelta(seconds=10)
        )
```

`@pegasus_workflow` wraps `temporalio.workflow.defn` and records the
`(name, version)` used by the manifest.

### Input contract: how `run()` receives its argument

Your `run()` method receives a **single positional argument** whose shape depends on how the workflow
was started:

**1. Trigger-fired (domain-event trigger)** — the dispatcher passes the full event envelope:

```python
{
    "domainEventId": "<uuid>",
    "eventType": "quote.accepted",      # the event type that fired the trigger
    "occurredAt": "<ISO-8601>",
    "payload": {"quoteId": "<id>", "moveId": "<id>"}   # entity ids, camelCase
}
```

Read entity ids from `arg["payload"]["quoteId"]` etc. The `payload` is a pointer, not a full snapshot
— always re-fetch authoritative state from the Pegasus API using those ids rather than relying on the
payload alone.

**2. Manual run** — `POST /api/v1/workflows/:id/run` passes:

```python
{"executionId": "<uuid>", "input": <user-supplied dict>}
```

Read your business data from `arg["input"]` (e.g. `arg["input"]["quote_id"]`).

**3. CLI test** — `pegasus-workflows test <name>` passes a raw string for local-dev parity.

Your `run()` should handle all three shapes. A module-level helper (not a method) is the recommended
pattern — it stays unit-testable without a Temporal worker context:

```python
def _resolve_quote_id(payload: dict | str) -> str:
    if isinstance(payload, str):
        return payload
    event_payload = payload.get("payload") if isinstance(payload, dict) else None
    if isinstance(event_payload, dict) and event_payload.get("quoteId"):
        return str(event_payload["quoteId"])
    inner = payload.get("input") if isinstance(payload, dict) else None
    if isinstance(inner, dict) and inner.get("quote_id"):
        return str(inner["quote_id"])
    return "quote-unknown"
```

### Sending an SMS

Inside a workflow activity, call `client.send_sms` to send an outbound text message
via the tenant's configured SMS provider. The platform holds the provider credentials —
no credential needs to appear in the workflow source or manifest.

```python
from pegasus_workflows import activity
from pegasus_workflows.api import PegasusClient

@activity.defn
async def send_alert_sms(to: str, message: str) -> dict:
    client = PegasusClient.from_runtime()   # reads the runner-injected env vars
    return client.send_sms(to=to, body=message)
```

> **Build the client with `PegasusClient.from_runtime()`** inside activities. The
> tenant runner injects the API connection as `PEGASUS_API_BASE_URL` and
> `PEGASUS_RUNTIME_TOKEN`; `from_runtime()` reads exactly those and raises a clear,
> named error if run outside the runner. Don't hardcode `os.environ[...]` — and
> note `PEGASUS_WORKFLOW_TOKEN` is the **publish-time CLI** token, _not_ a runtime
> var, so reaching for it here fails at runtime.

Declare the capability in `pegasus-workflows.toml`:

```toml
[[workflow]]
name = "order-saved-notify"
version = "0.1.0"
entry_points = ["order_saved.workflow:OrderSavedWorkflow"]
required_actions = ["SendSms"]
```

`send_sms` raises `PegasusApiError` (403) if `SendSms` is absent from `required_actions`,
or (404) if the tenant has no SMS provider connected. The `to` number must be E.164
(e.g. `"+16308868537"`).

### Soliciting feedback (magic-link surveys)

Ask a customer or driver for feedback via a tokenized link, then act on the
response. You author a versioned form once (via the `feedback-form` CLI), then a
workflow **mints a per-recipient link** and sends it. When the recipient submits,
the platform records the response and emits the built-in `feedback.submitted`
domain event — so a second workflow with an `EVENT` trigger on `feedback.submitted`
picks up and routes it (alert on a low rating, log a testimonial, etc.).

Author the form (once), from a working directory holding `form.json`:

```json
{
  "title": "How did we do?",
  "definition": {
    "questions": [
      { "id": "rating", "type": "rating", "label": "Rate your crew", "required": true },
      { "id": "comments", "type": "text", "label": "Anything else?", "maxLength": 500 }
    ]
  }
}
```

```bash
pegasus-workflows feedback-form validate post-move-csat   # dry-run the definition
pegasus-workflows feedback-form publish  post-move-csat   # publish v1
```

Then, inside a workflow activity, mint + send the link:

```python
@activity.defn
async def request_feedback_activity(move: dict) -> dict:
    client = PegasusClient.from_runtime()
    # channel="sms" also sends the form's messageTemplate; omit it to mint-only
    # and send the returned url yourself (e.g. via client.send_sms or email).
    return client.create_feedback_request(
        "post-move-csat",
        subject_type="move",
        subject_id=move["id"],
        ttl_hours=72,
        channel="sms",
        to=move["contactPhone"],
    )
```

`create_feedback_request` needs `required_actions = ["CreateFeedbackRequest"]`;
form authoring (`publish_feedback_form`, …) needs `ManageFeedbackForms`. Question
types: `rating` (int, default 1..5), `number`, `text`, `select` (options), `boolean`.
Poll a request with `client.get_feedback_request(request_id)`. The whole feature is
server-gated behind `FEEDBACK_ENABLED`.

### Delivering a body to a partner endpoint

To POST a mapped body to a partner API, use `client.deliver_to_external` rather than
calling the partner directly with `httpx` — the platform performs the outbound POST
**server-side**, so the send flows through the one boundary a dry run controls
(captured, never performed) instead of a raw call the runtime can't see or stop. It
pairs with `map_to_external`: map to build the body, deliver to send it.

```python
@activity.defn
async def send_order_to_partner(order: dict) -> dict:
    client = PegasusClient.from_runtime()
    mapped = client.map_to_external("demo_partner", order)   # build the partner body
    if not mapped["valid"]:
        raise RuntimeError(f"refusing to send invalid body: {mapped['issues']}")
    result = client.deliver_to_external("demo_partner", mapped["external"])
    if not result["delivered"]:
        raise RuntimeError(f"partner rejected the delivery: {result['status']}")
    return result
```

The delivery URL and API key come from the workflow's own config/secret (`SEND_URL`
config, `SEND_API_KEY` secret by default — override with `url_config` /
`api_key_secret` / `headers_config` / `group`), so no partner URL or key appears in
the workflow source. Declare `required_actions = ["DeliverToExternal"]` in the
manifest. Returns `{delivered, status, response, dryRun}`; raises `PegasusApiError`
on 403 (missing action), 404 (unknown integration, or the URL config / API-key
secret is not set), or 400 (a delivery URL pointing at a private/loopback host).

**It is one fixed request, on purpose.** A single JSON `POST` to the whole URL in
`SEND_URL`, always sent with `Authorization: Bearer <SEND_API_KEY>` — no
per-request path, and the bearer isn't optional (no `SEND_API_KEY`, no delivery).
That narrowness is what makes it two config entries and one line of workflow code.
Use [`call_external`](#calling-a-partner-api-authenticated-reads--writes) instead
as soon as you need a per-request path, a non-bearer credential (a named header /
`AUTH_MODE=apikey`, OAuth2, or `none`), or more than one credential header —
`call_external(..., method="POST", path=..., body=...)` is the same server-side,
dry-run-capturable boundary with those knobs exposed, so the guarantees are
identical. Not a drop-in swap of the call alone, though: it wants
`required_actions = ["CallExternal"]`, and it builds the endpoint from `BASE_URL`

- `path` with the credential picked by `AUTH_MODE` rather than from `SEND_URL` +
  `SEND_API_KEY`. Delivery for the common case, `call_external` for anything past it.

So the tenant can see which keys to provision, **declare them on the integration
config** via `required_secrets` / `required_configs` (each `{key, group?,
description?}`) — e.g. `required_secrets=[{"key": "SEND_API_KEY", "group": "sirva"}]`,
`required_configs=[{"key": "SEND_URL", "group": "sirva"}]` on
`publish_integration_config`, or as `requiredSecrets`/`requiredConfigs` in the
config directory's `meta.json`. It is informational (the runtime read still resolves
lazily) but surfaces a present/missing view on the integration's detail page and the
Settings → Developer → Configs summary. Read the resolved status back with
`pegasus-workflows requirements --missing-only` or
`PegasusClient.integration_requirements_summary()` (presence only — never values).

### Ingesting a partner payload (inbound: native → canonical)

`map_to_external` is the **outbound** direction (entity → partner body). To go the
other way — normalize a partner's **native** payload into the platform's canonical
entity — use `client.map_from_external`. This is the step an **ingest workflow**
runs on an inbound webhook event (see "Inbound integration ingress"): map the raw
partner payload to the canonical shape, then persist it (e.g. to a projection).

```python
@activity.defn
async def normalize_shipment_event(event: dict) -> dict:
    client = PegasusClient.from_runtime()
    result = client.map_from_external("sirva_ade_shipment", event)   # native → canonical
    if result["canonical"] is None or not result["valid"]:
        raise RuntimeError(f"refusing to ingest an invalid payload: {result['issues']}")
    return result["canonical"]   # the system-of-record entity to persist
```

Returns `{canonical, valid, issues, degraded}`. `canonical` is the normalized
entity (the value the outbound direction discards internally) — `None` only when
the payload can't be mapped/parsed at all, so an ingest can **fail closed**.
`valid`/`issues` are the same gate verdict. Raises `PegasusApiError` (404) on an
unknown integration / no floor — it fails closed so an ingest never proceeds on a
silently-empty entity. Open API-key surface — no `required_actions`.

The integration's mapping + rules are a **published config** on a reusable,
partner-neutral **floor** (`shipment_lifecycle_event`, `sales_lead`,
`financial_settlement`, `document_record`). Partner value sets (allowed brand
codes, statuses, file types) live in the config's rules via the `nin` operator
(`{brandPresent eq true} AND {brand nin [AVL,NVL]}`) — not in platform code — so
the same floor serves any partner of that type.

### Calling a partner API (authenticated reads & writes)

`deliver_to_external` is one fixed JSON `POST`. For arbitrary reads and writes —
`GET` with query params, per-call paths, OAuth2 partners — use
`client.call_external`. You name a `method` + `path` (+ `query`/`body`) and the
platform performs the call **server-side** against the integration's configured
`BASE_URL`, authenticating per its `AUTH_MODE`. For `oauth2_client_credentials` it
mints, caches, and re-mints (on a partner `401`) an OAuth2 token server-side — so
`client_id`/`client_secret` never appear in workflow code.

```python
@activity.defn
async def fetch_shipment(reg: str, year: int) -> dict:
    client = PegasusClient.from_runtime()
    res = client.call_external(
        "sirva_ade_shipment",
        method="GET",
        path="/OM/m1/GetShipmentDetail",
        query={"RegNumber": reg, "RegYear": year},
    )
    return res["response"]          # parsed JSON (or raw text for an XML reply)
```

Config + credentials live in the tenant's config/secret store, read by name +
`group`. Declare `required_actions = ["CallExternal"]`. Returns
`{status, ok, response, headers, attempts, dryRun}`, where `headers` is **every**
partner response header (lowercase keys, `set-cookie` removed) — so `retry-after`,
`etag`, and vendor diagnostics like `x-ms-request-id` are all readable.

| `AUTH_MODE`                           | Credential                                                | Sent as                                                                          |
| ------------------------------------- | --------------------------------------------------------- | -------------------------------------------------------------------------------- |
| `oauth2_client_credentials` (default) | `CLIENT_ID` + `CLIENT_SECRET` secrets, `TOKEN_URL` config | minted bearer, cached, re-minted on `401`                                        |
| `bearer`                              | `API_KEY` secret                                          | `Authorization: Bearer …`                                                        |
| `apikey`                              | `API_KEY` secret                                          | the header named by config `API_KEY_HEADER`, default `Ocp-Apim-Subscription-Key` |
| `none`                                | —                                                         | —                                                                                |

**Dry-run split:** a `GET` is a read and runs **live** under
`pegasus-workflows run --dry-run` (returns real data); a `POST`/`PUT`/… is a
mutation and is **captured, not performed**. Pass `mutating=True`/`False` to
override the method-based default when a partner overloads a verb.

#### Custom request headers

Two maps, split by **trust level** — this split is the point, so pick the right one:

- `headers` — literal values, **non-secret by construction** (they come from your
  workflow code): `{"On-Behalf-Of": "jdoe"}`.
- `secret_headers` — header name → **secret key name**. The platform resolves the
  value from the encrypted store, so the credential never appears in workflow
  source, logs, or a captured dry-run payload:
  `{"X-Partner-Token": "PARTNER_TOKEN"}`.

`Authorization`, `Host`, `Content-Length` and `Content-Type` are owned by the
platform and rejected with a `400` — allowing an override would let a workflow
bypass `AUTH_MODE` entirely. Header names must be RFC 7230 tokens and values may
not contain CR/LF.

An Azure API Management partner (subscription key + a per-request identity
header) looks like this — with tenant config `AUTH_MODE=apikey` and secret
`API_KEY` holding the subscription key:

```python
res = client.call_external(
    "atlas_estimating",
    method="GET",
    path="/Estimating/Order/12345",
    headers={"On-Behalf-Of": "jdoe"},   # identity, not a credential
)
```

#### Timeouts and throttling

Each attempt is bounded by config `REQUEST_TIMEOUT_MS` (default 30s, clamped to
[1000, 60000]); exceeding it raises a `504`. A `429`/`503` is retried up to config
`MAX_RETRIES` times (default 2, clamped to [0, 5]), honoring the partner's
`Retry-After` (capped at 10s) and otherwise backing off exponentially.

**Only idempotent requests are retried** — `GET`/`HEAD`/`OPTIONS`, or any call
with `mutating=False`. A `POST` is never auto-retried, because a repeat could
double-write at the partner. Read `attempts` in the result to see how many HTTP
requests were actually made.

`deliver_to_external` takes the same `headers` / `secret_headers` /
`timeout_config` arguments. Its older `headers_config` (a config key holding a
JSON object) still works but is **non-secret only** — config values are stored in
plaintext, so a credential belongs in `secret_headers`.

**A partner needing two credential headers** — a subscription key plus a client
id, say — is `secret_headers` with two entries, not a new `AUTH_MODE`. Auth modes
name _one_ platform-managed credential scheme; a second header is caller-specific,
so it belongs to the caller:

```python
client.call_external(
    "some_partner",
    method="POST",
    path="/orders",
    body=payload,
    secret_headers={
        "Ocp-Apim-Subscription-Key": "PARTNER_SUB_KEY",
        "X-Client-Id": "PARTNER_CLIENT_ID",
    },
)
```

with `AUTH_MODE=none` if neither header is the integration's configured auth, or
`AUTH_MODE=apikey` if one of them is (then name only the _other_ here). The values
are secret **key names**, resolved server-side from the encrypted store — the
credential itself never enters workflow code, a dry-run capture, or a log.

### Transferring documents (blobs)

A workflow can stage a file to upload or land a file it fetched without holding
the bytes in workflow memory — `put_blob`/`get_blob` stream **runner↔S3 directly**
(presigned URLs), so they aren't bounded by the API payload limit. Declare
`required_actions = ["WriteBlob"]` (put) / `["ReadBlob"]` (get).

```python
@activity.defn
async def file_document(reg: str) -> dict:
    client = PegasusClient.from_runtime()

    # Retrieve: land a partner GET response straight into a blob (not memory).
    got = client.call_external(
        "sirva_ade_document", method="GET", path="/IMAGING/m2/GetImage",
        query={"Id": reg}, response_to_blob=True,
    )
    blob_id = got["blobId"]

    # ...or stage your own bytes:
    handle = client.put_blob(pdf_bytes, content_type="application/pdf")  # {blobId, size}

    # Upload: reference a staged blob; the platform inlines its bytes server-side.
    client.call_external(
        "sirva_ade_document", method="POST", path="/Imaging/m3/AddDocument",
        body={"ReferenceNumber": reg, "FileData": {"$blob": handle["blobId"]}},
    )
    return {"blobId": blob_id}
```

`put_blob` is a mutation (captured under `--dry-run`); `get_blob`/`get_blob_url`
are reads (live). Blobs are tenant-scoped and expire via a TTL. The
`$blob`/`response_to_blob` paths are a **small-file cut** (≤ ~5 MB through the
API); large-file streaming is a follow-up.

### Receiving events (inbound ingress)

For partners that **push** to you (a webhook), provision a platform-hosted ingress
endpoint. The partner POSTs to it with a bearer the platform issues; the endpoint
authenticates, dedups, persists the raw body, emits a domain event, and returns a
partner-shaped ack synchronously.

```
pegasus-workflows ingress create sirva_ade_shipment   # prints URL + one-time token
pegasus-workflows ingress rotate sirva_ade_shipment
pegasus-workflows ingress list   sirva_ade_shipment
```

The workflow that handles the events binds to the emitted domain event with an
ordinary **EVENT trigger**. The emitted event type, the dedup key path, the body
**`validation`**, and the **ack template** (the partner's `Result{…}` envelope)
are published as part of the integration definition — an `inbound` block on the
integration config:

```jsonc
"inbound": {
  "eventType": "sirva_ade.shipment.event",
  "dedupKeyPath": "Events.0.Id",
  // Declarative body checks. A body that fails them gets the `failure` ack (below)
  // at HTTP 200 — the partner records the rejection and does not retry it.
  "validation": {
    "requiredPaths": ["SvcProvDataRecipient"],
    "nonEmptyArrayPaths": ["Events"]
  },
  "ackTemplate": {
    // {{key}} substitutes a whole context value ({{status}} → "Success"/"Failed",
    // {{errorCount}} → the issue count). A number stays a number.
    "success": { "Result": { "Results": "{{status}}", "ResultsMessageCount": "{{errorCount}}",
                             "ResultsMessage": [] } },
    // The `$map` directive builds a per-message array from the structured issues —
    // exactly ADE's ResultsMessage[{Code, Description}] shape. Each element renders
    // the `as` sub-template against one issue ({{code}}, {{message}}).
    "failure": { "Result": { "Results": "{{status}}", "ResultsMessageCount": "{{errorCount}}",
      "ResultsMessage": { "$map": "issues",
        "as": { "ResultsMessageCode": "{{code}}", "ResultsMessageDescription": "{{message}}" } } } }
  }
}
```

The ack is derived from **ingestion** (accepted + durably queued), never from the
bound workflow finishing. Managing ingress needs a `vnd_` key with `ManageIngress`
(the `workflow_developer` / `tenant_admin` role). With no `inbound` block, the
endpoint accepts any body and returns a generic `{status:"accepted"}` ack (no
validation, no partner envelope).

**Multi-shape partners — `validation.oneOf` (0.22.0+).** When one ingress id
receives structurally different bodies (e.g. an ADE Abstract `{…, AgentNbr,
StatementEntry[]}` vs a Statement `{AgentStatementHdr:{AgentNbr}, PostingTickets[]}`),
list each accepted shape under `validation.oneOf`. The body must fully satisfy **at
least one** variant (in addition to any top-level `requiredPaths`/`nonEmptyArrayPaths`);
a body matching none gets the `failure` ack. Pair it with an array `dedupKeyPath`
(each path tried in order, first present wins) so both shapes dedup:

```jsonc
"inbound": {
  "eventType": "sirva_ade.compensation.event",
  "dedupKeyPath": ["StatementEntry.0.ReferenceNbr", "PostingTickets.0.ReferenceNbr"],
  "validation": {
    "oneOf": [
      { "requiredPaths": ["AgentNbr"],                  "nonEmptyArrayPaths": ["StatementEntry"] },
      { "requiredPaths": ["AgentStatementHdr.AgentNbr"], "nonEmptyArrayPaths": ["PostingTickets"] }
    ]
  }
  // ackTemplate as above
}
```

### Secrets & configuration

A workflow reads two kinds of per-tenant key/value data at runtime — **secrets**
(write-once, encrypted at rest; e.g. a third-party API key) and **config** (plain,
editable; e.g. a region or base URL). Both are scoped to the whole tenant, so every
workflow the tenant owns reads the same namespace. Values live in the platform — they
never appear in the workflow source or artifact.

**1. Publish the values once** with a `vnd_` key holding the manage actions (the
`workflow_developer` or `tenant_admin` role), via the CLI:

```
pegasus-workflows secrets set STRIPE_API_KEY "sk_live_…" --token=vnd_… --base-url=…
pegasus-workflows config  set DEFAULT_REGION us-east-1   --token=vnd_… --base-url=…
pegasus-workflows secrets list --token=vnd_…   # metadata only — never values
```

or from Python (`client.set_secret(...)`, `client.set_config(...)`,
`client.list_secrets()`, `client.delete_secret(...)`). Secrets are write-once —
delete then set again to rotate; `set_config` is an idempotent upsert.

**2. Declare the read actions** your workflow needs in `pegasus-workflows.toml`,
and — recommended — the specific keys it reads, so the tenant sees up front which
values to provide and whether they are set:

```toml
[[workflow]]
name = "charge-on-quote-accepted"
version = "0.1.0"
entry_points = ["charge.workflow:ChargeWorkflow"]
required_actions = ["ReadWorkflowSecret", "ReadWorkflowConfig"]
# Which keys this workflow reads. Each is a table with a required `key`, an
# optional `group` (default "global"), and an optional `description`. Purely
# informational — it does not gate execution — but it drives the tenant UI's
# "keys still needed" view in Settings → Developer → Configs and the badges on
# the workflow's detail page.
required_secrets = [{ key = "STRIPE_API_KEY", group = "billing", description = "Stripe secret key" }]
required_configs = [{ key = "DEFAULT_REGION" }]
```

To see which of those keys the tenant has actually set, read the resolved state
back — presence only, never values:

```bash
pegasus-workflows requirements --missing-only --token vnd_... --base-url https://api...
```

`MISSING  billing  STRIPE_API_KEY  secret  workflow:send-quote-followup  Stripe secret key`

or from code, `PegasusClient.requirements_summary()`. Both planes are merged by
the command; if your token can read only one of them it says so rather than
reporting a partial answer as complete.

**3. Read the values inside an activity** (never in workflow code):

```python
@activity.defn
async def charge_customer(amount_cents: int) -> str:
    client = PegasusClient.from_runtime()
    api_key = client.get_secret("STRIPE_API_KEY")   # needs ReadWorkflowSecret
    region = client.get_config("DEFAULT_REGION")    # needs ReadWorkflowConfig
    ...
```

`get_secret` / `get_config` raise `PegasusApiError` (404) if the key is unset and
(403) if the matching read action is absent from `required_actions`.

### Integration projections (cached external state)

When a workflow syncs an external system, it can cache each record's last-known
state as a **projection** — keyed by `(integration, entity_type, key)` within the
tenant. The Pegasus **integration validator** reads the matching record's cached
`state` back as the `prior` input when pre-validating an update, so transition
rules stay accurate without the caller resupplying prior state.

Declare the actions your workflow needs, then read/write inside an activity:

```toml
[[workflow]]
name = "sync-demo-partner-orders"
version = "0.1.0"
entry_points = ["sync.workflow:SyncDemoPartner"]
required_actions = ["ReadIntegrationProjection", "WriteIntegrationProjection"]
```

```python
@activity.defn
async def cache_order(order: dict) -> None:
    client = PegasusClient.from_runtime()
    # Mirror the external record (native payload shape, ≤ 256 KB serialized).
    client.put_projection("demo_partner", "order", order["serviceOrderNumber"], order)

    prior = client.get_projection("demo_partner", "order", "SO-12345")  # None on miss
    every = client.list_projections("demo_partner", "order")
    client.delete_projection("demo_partner", "order", "SO-12345")
```

`get_projection` returns `None` on a cache miss; the write methods raise
`PegasusApiError` (403) if the matching action is absent from `required_actions`,
and `put_projection` raises 413 if the serialized state exceeds 256 KB.

### Reading operational entities (inside a workflow)

A running workflow authenticates with its `workflow_runtime` service-account key
(`PegasusClient.from_runtime()`), which is granted read access to the core
operational records. These helpers return `{data, meta: {total, count, limit,
offset}}` and take `limit` (≤100) / `offset`:

```python
client = PegasusClient.from_runtime()
client.list_customers(limit=25)          # ReadCustomer
client.list_quotes()                     # ReadQuote
client.list_moves()                      # ReadMove
client.list_invoices()                   # ReadInvoice
client.list_events("order.completed")    # ReadEvent — poll pending inbound events of a type
```

They read the m2m `/api/v1/runtime/*` surface (the browser `/api/v1/*` CRUD routes
are Cognito-only and reject a `vnd_` key). `list_events` polls the inbound
platform-event queue, which is keyed by type, so an event type is required.
(Inventory has no runtime read grant — a workflow that needs item-level data reads
it from the move it is processing.)

The pegII operational surface (legacy orders + tasks + salesmen) has its own reads/mutation:

```python
client.list_orders()                 # ReadOrder
client.get_order("SO-12345")         # ReadOrder — projected row {id, orderNumber, status, …}
client.list_tasks(order_id="SO-12345")   # ReadTask
client.get_task("task-1")            # ReadTask
client.close_task(order_id="SO-12345", task_type="date_confirmation", reason="done")  # CloseTask
client.list_salesmen(active="true")  # ReadSalesman
client.get_salesman("213056")        # ReadSalesman — {id, name, email, branch, active, …}
```

To dry-run a **published integration** against a real order id — "does this
production order pass the mapping?" — fetch the order's **native** pegII payload
(the same `{Id, Survey, InvolvedParties, KeyMoveDates, …}` shape a partner posts to
the ingress) and normalize it through the integration's inbound mapping. Two ways:

```python
# One-shot: fetch native + map + gate, server-normalized.
client.dry_run_integration("demo_partner", "490574")   # ReadOrder
# → {canonical, valid, issues, degraded}

# Or compose it yourself from the native payload:
native = client.get_order("490574", shape="native")    # ReadOrder — raw pegII payload
client.map_from_external("demo_partner", native)        # native → canonical + gate
```

No hand-pasting the raw payload. `get_order` without `shape` returns the projected
row; `shape="native"` returns the raw serialized object for mapping.

### Emitting a custom event (workflow-to-workflow chaining)

A workflow can fire a tenant-defined event type, which any workflow bound to it via
an EVENT trigger will run — the in-platform way to chain automations without an
outside queue. Requires `EmitTenantEvent`.

```python
client.emit_event("quote.followed_up", {"quoteId": "Q-1", "channel": "sms"})
# → {"emitted": True, "eventType": "quote.followed_up", "occurredAt": "…"}
```

The event type must already exist for the tenant (define it in the tenant UI or via
the events surface). `emit_event` is a mutation — captured, not performed, under the
offline test harness and the server-side `--dry-run`.

### Calling a read endpoint directly (`api_get`)

Most reads have a typed helper, but the API exposes more than the helpers cover —
for example the **projection read-model** (`GET /integrations/{id}/projections/{entityType}`),
which the typed `get_projection` / `list_projections` don't reach: it filters and
keyset-pages. `api_get(path, **params)` is a **read-only** passthrough to any Pegasus
API path with your key; it returns the full JSON body (so `meta` / `nextCursor`
survive). The catalog of paths is the OpenAPI spec (`GET /openapi.json`, or
`pegasus://reference/openapi` in your agent):

```python
# "ADE shipments still stuck at REGISTERED, not touched since 3 days ago" — paged
page = client.api_get(
    "/api/v1/integrations/sirva_ade_shipment/projections/shipment",
    status="REGISTERED", updatedSince="2026-07-14T00:00:00Z", limit=50,
)
for record in page["data"]:
    ...                       # chase the stuck shipment
cursor = page.get("nextCursor")   # keyset-page through the rest
```

Read-only by design: `api_get` only does `GET`, and only against your Pegasus API
(an absolute URL raises — a partner host is `call_external`'s job). For **writes**,
use the typed methods — they route through the dry-run capture path, which a generic
call would bypass. `api_get` is likewise **not** stubbed by the offline test harness
(use a typed read helper there); it's meant for ops/reconciliation on a real client.

## Visualizing workflows

A workflow is published as opaque Python, so the Pegasus tenant UI can't infer
what it does. Instead, each workflow ships a **Mermaid diagram** at
`<source_dir>/workflow.mmd` that business users view to confirm the workflow
matches their business rules. The UI pairs it with a _verified envelope_ drawn
from data the platform actually stores and trusts — the workflow's triggers, its
declared `required_actions`, and the secret/config keys it touches — so the
diagram (author-declared) sits next to the permission boundary (platform-guaranteed).

A diagram is **required to publish** — but how you draw it is up to you. The
`workflow.mmd` file is the source of truth: hand-write it, or have it drawn by
**whatever coding agent you already use** (Claude Code, Cursor, Copilot, …) on
your own subscription. `init` scaffolds a starter `workflow.mmd` so a new project
publishes out of the box.

The `diagram` command never calls an LLM and needs no API key. It assembles a
ready-to-use prompt — your workflow's Python source plus the exact output path and
formatting rules — for your agent to act on:

```
pegasus-workflows diagram                  # print the prompt for all workflows
pegasus-workflows diagram -w my-workflow   # just one workflow
pegasus-workflows diagram -o diagram.txt   # write the prompt to a file instead
```

Feed the output to your coding agent (it names the target path, e.g.
`<source_dir>/workflow.mmd`, and asks for a bare `flowchart TD`), then save the
result there. The file is packaged into the bundle, so it is SHA-pinned to the
exact published version — **edit it freely**; a changed diagram only goes live
with a new published version.

## Inspecting executions

Read execution status, results, and the Temporal event-history timeline from the
terminal (the same tenant-scoped data the web UI shows):

```
pegasus-workflows executions list <workflow-id> --token=vnd_…
pegasus-workflows executions show <workflow-id> <execution-id> --token=vnd_…
```

`show` prints the run's input/result/error plus a flattened timeline
(`WorkflowExecutionStarted` → per-activity events → the terminal event). The same
data is available programmatically via `client.list_executions`,
`client.get_execution`, and `client.get_execution_history`.

Drive an execution from the SDK too (also in the tenant web UI):

```python
client.cancel_execution(workflow_id, execution_id)  # CancelWorkflowExecution — cooperative cancel
client.retry_execution(workflow_id, execution_id)   # RetryWorkflowExecution — new run, same input
```

`cancel` signals a running execution (it transitions to CANCELLED when the run
observes it); `retry` starts a **new** execution from the stored input of a
terminal-failed one (FAILED / TIMED_OUT / CANCELLED), leaving the original row
untouched. Both need the `workflow_developer` role.

> ⚠️ **Keep PII out of workflow inputs and results.** Temporal stores execution
> payloads (input, result, and the full event history) and renders them in its
> UI, and platform engineers can read them cross-tenant in the Temporal Cloud
> console. Pass **entity ids**, not raw personal data — look the details up inside
> an activity via the API. (A payload codec would let us encrypt payloads; it's
> deferred until this convention can't hold.)

## The manifest — `pegasus-workflows.toml`

Every project has a `pegasus-workflows.toml` at its root. Each `[[workflow]]`
table is packaged into its own artifact and uploaded as a distinct
`(name, version)` row:

```toml
[[workflow]]
name = "demo"                                   # ^[a-z0-9][a-z0-9_-]{0,63}$
version = "0.1.0"                               # semver
entry_points = ["demo.workflow:HelloWorkflow"]  # non-empty
source_dir = "demo"                             # optional, defaults to name
description = "..."                             # optional
```

These rules mirror the server's `ManifestSchema` exactly, so `package`/`push`
fail fast locally before any HTTP call.

## Credentials & profiles

Every command that talks to the API needs a `vnd_` token and a base URL. Rather
than pasting them on the command line (where they leak into shell history,
process listings, and agent transcripts), store **named profiles** — AWS-CLI
style — in `~/.pegasus/credentials`:

```
pegasus-workflows configure --profile prod   # prompts for api_key (hidden) + api_root
pegasus-workflows configure --profile qa
pegasus-workflows profile list               # names + api_root only — never the key
```

The file is created `0600` (owner read/write only) and is **never committed** —
keep it out of repos. Then select a profile per command:

```
pegasus-workflows push --profile prod        # token + root from [prod]
pegasus-workflows run  --profile qa <id>
pegasus-workflows push                        # uses [default] if present
```

`--profile` works on every command that builds a client (`push`, `run`,
`integration-config`, `executions`, `secrets`, `config`). Resolution precedence,
highest first:

| Tier | Source                                                 |
| ---- | ------------------------------------------------------ |
| 1    | explicit `--token` / `--base-url` flags                |
| 2    | `--profile NAME`                                       |
| 3    | `PEGASUS_WORKFLOW_TOKEN` / `PEGASUS_BASE_URL` env vars |
| 4    | the `[default]` profile                                |

`api_root` is optional in a profile and defaults to
`https://api.pegasus.dolas.dev`. With nothing configured at all, the base URL
falls back to `http://localhost:3000` for local dev.

> These are the **publish-time CLI** credentials — distinct from the runtime vars
> (`PEGASUS_API_BASE_URL` / `PEGASUS_RUNTIME_TOKEN`) that the tenant runner injects
> for `PegasusClient.from_runtime()` inside activities.

## Deployment ledger — `deployments.toml`

Workflow ids are **environment-specific** — publishing the same workflow to QA and
prod yields different ids. After a successful `push`, the SDK records where each
workflow landed in a `deployments.toml` beside the manifest, so post-publish
actions (`run`, `executions`, fork, rollback) read the id instead of scraping it
from scrollback:

```toml
[prod]
base_url = "https://api.pegasus.dolas.dev"
workflow_id = "f8077342-2e58-4dc1-a47a-797ca394ef72"
version = "0.1.0"
visibility = "GLOBAL"
published_at = "2026-06-29T21:05:48Z"
```

- The environment key is derived from the API host, or set explicitly with
  `push --env NAME`.
- Re-publishing to the same env **updates the entry in place** (no duplicates);
  a second env adds a table. A multi-workflow project nests each record under the
  workflow name (`[prod.send_order_saved_sms]`).
- The file is **safe to commit** — it holds ids and URLs only, never a token.

## CLI

| Command                                                                            | What it does                                                   |
| ---------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| `pegasus-workflows setup [--profile <name>] [--print-mcp-config]`                  | First-run bootstrap: seed a profile + register the MCP server. |
| `pegasus-workflows init <name>`                                                    | Scaffold a new workflow project.                               |
| `pegasus-workflows configure [--profile <name>]`                                   | Store a credential profile in `~/.pegasus/credentials` (0600). |
| `pegasus-workflows profile list`                                                   | List stored profile names + api_root (never the key).          |
| `pegasus-workflows diagram [-C <dir>] [-w <name>] [-o <file>]`                     | Print a prompt for your coding agent to draw `workflow.mmd`.   |
| `pegasus-workflows package`                                                        | Zip each declared workflow into `dist/<name>-<version>.zip`.   |
| `pegasus-workflows push [--profile <name>] [--env <name>] [--token=…]`             | Package → upload → finalize; records `deployments.toml`.       |
| `pegasus-workflows test <workflow>`                                                | Start local Temporal and run the workflow with a stub input.   |
| `pegasus-workflows executions list <wf-id> --token=<vnd_…>`                        | List recent executions of a workflow (newest first).           |
| `pegasus-workflows executions show <wf-id> <exec-id> --token=<vnd_…>`              | Show one execution's input/result/error + history timeline.    |
| `pegasus-workflows integration-config validate <id> [-C <dir>]`                    | Dry-run the publish gate for a config (no write).              |
| `pegasus-workflows integration-config publish <id> [-C <dir>]`                     | Gate then publish a new config version.                        |
| `pegasus-workflows integration-config pull <id> [-C <dir>] [--stdout]`             | Fetch the active config; write the editable surface to disk.   |
| `pegasus-workflows integration-config versions <id>`                               | List the config version history (newest first).                |
| `pegasus-workflows integration-config rollback <id> <version>`                     | Re-publish a prior version (re-runs the gate).                 |
| `pegasus-workflows integration-config fork <id> [--force] [--yes]`                 | Copy the GLOBAL config into this tenant; `--force` re-syncs.   |
| `pegasus-workflows integration-config delete <id> [--force] [--yes]`               | Permanently remove the caller's config for an integration.     |
| `pegasus-workflows secrets set <key> <value> [-d <desc>]`                          | Publish a secret (write-once, encrypted at rest).              |
| `pegasus-workflows secrets list` / `secrets delete <key>`                          | List secret keys (no values) / delete a secret.                |
| `pegasus-workflows config set <key> <value> [-d <desc>]`                           | Publish a config value (idempotent upsert).                    |
| `pegasus-workflows config list` / `config delete <key>`                            | List config key/values / delete a config entry.                |
| `pegasus-workflows requirements [--missing-only] [--json]`                         | Which declared secret/config keys are set, and what's missing. |
| `pegasus-workflows schedule create <wf-id> --cron "<5-field UTC>"`                 | Attach a cron SCHEDULE trigger that runs the workflow.         |
| `pegasus-workflows schedule list <wf-id>` / `schedule delete <wf-id> <trigger-id>` | List / remove a workflow's schedule triggers.                  |
| `pegasus-workflows ingress create <id>` / `rotate <id>` / `list <id>`              | Provision / rotate / inspect a partner-ingress bearer.         |
| `pegasus-workflows feedback-form validate <key> [-d <dir>]`                        | Dry-run a feedback form definition (no write).                 |
| `pegasus-workflows feedback-form publish <key> [-d <dir>]`                         | Publish a new feedback form version.                           |
| `pegasus-workflows feedback-form pull <key> [-d <dir>]`                            | Fetch the active form; write `form.json` (+ `message.txt`).    |
| `pegasus-workflows feedback-form versions <key>` / `rollback <key> <version>`      | List versions / re-publish a prior version.                    |

Credentials resolve via `--token`/`--base-url`, `--profile`, the
`PEGASUS_WORKFLOW_TOKEN`/`PEGASUS_BASE_URL` env vars, or the `[default]` profile
(see [Credentials & profiles](#credentials--profiles)). The token is a `vnd_*`
Pegasus API key whose service account holds the `workflow_developer` role.

### Authoring an integration-validator config

The `integration-config` group manages an integration's declarative **mapping +
rules** (the DB-backed authoring surface). The working directory (`-C`, default
`.`) holds:

| File                                            | Required | What it is                                                                                                                             |
| ----------------------------------------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------- |
| `mapping.json`                                  | ✓        | native payload → canonical (output-shaped DSL; `GET /api/v1/integrations/mapping-schema`)                                              |
| `rules.json`                                    | ✓        | behavioral rules; ops `eq ne gt gte lt lte in nin` (use `nin` for "must be one of a set")                                              |
| `corpus.json`                                   | ✓        | gate cases `{input: {order: <native>}, expected: {valid, ruleIds}}`                                                                    |
| `meta.json`                                     | –        | `{floor, displayName}` — **required for a new id** with no built-in overlay                                                            |
| `inbound.json`                                  | –        | ingress ack/validation block (`GET /api/v1/integrations/inbound-schema`) — makes the ingress return the partner's `Result{…}` envelope |
| `external-shape.json` / `external-mapping.json` | –        | partner **outbound** body shape + projection                                                                                           |

**Reformatting a date is a `coerce`, not a workaround.** Partners document their own
date format while the native payload carries .NET datetimes
(`2026-07-16T00:00:00`), so `mapping.json` has two date coercions —
`toDateOnly` → `YYYY-MM-DD` and `toIsoDateTime` → `YYYY-MM-DDTHH:mm:ss`. Both are
wall-clock truncations (a trailing `Z`/offset is dropped, never applied, so the day
cannot shift) and both are null-safe: `null`, an absent path, `""`, or an
unparseable date yields `null`. `coerce` runs **after** `$map`, so one leaf handles
the .NET min-date sentinel and the format together:

```jsonc
"surveyDate": {
  "$from": "KeyMoveDates.Survey.Planned",
  "$map": {"0001-01-01T00:00:00": null},  // sentinel → null …
  "coerce": "toDateOnly"                  // … every real date → YYYY-MM-DD
}
```

Nothing in the gate compares your output against the _partner's_ documented format
— the canonical contract types these fields as generic dates — so the format is
yours to get right. `client.get_mapping_schema()` is the live vocabulary (each
directive carries a `description`).

**Discover the floor first.** A floor is the contract your config builds on — it
declares the only legal mapping _targets_ and rule _facts_. Author against it:

```python
client.list_floors()                    # [{floor, canonicalFields, factCatalog, factDocs?, …}]
client.get_floor("shipment_lifecycle_event")
#  → canonicalFields:  legal mapping targets    (what a mapping may WRITE)
#  → factCatalog:      legal rule facts         (name → type)
#  → factDocs:         what each fact MEANS     (name → one line), when documented
#  → inputFieldRoots:  legal mapping source roots (what a $from may READ), when declared.
#      bare "Survey" opens a whole native root; dotted "UnusedFields.survey_received"
#      opens ONLY that curated sub-path (siblings stay closed).
```

**Read `factDocs` before picking a fact.** A name and a type don't say what a fact
counts. Facts that count related records (`shipmentsWith…`) carry "at least one"
semantics, so you forbid the milestone with `{"op": "lte", "value": 0}` — and two
such predicates AND-ed in one rule are evaluated **independently**, so with 2+
related records one may satisfy the first and a different one the second. Where the
values must belong to the _same_ record the floor publishes a paired fact; `factDocs`
is what tells you which is which:

```jsonc
// "at least one shipment has a load actual" — the load-only milestone
"when": [
  {"fact": "serviceStatus", "op": "eq",  "value": "In Progress"},
  {"fact": "shipmentsWithLoadActual", "op": "lte", "value": 0}
]

// load AND delivery on the SAME shipment — use the paired fact, not two predicates
"when": [
  {"fact": "serviceStatus", "op": "in",  "value": ["Delivered", "Completed"]},
  {"fact": "shipmentsWithLoadDeliveryActual", "op": "lte", "value": 0}
]
```

(or public `GET /api/v1/integrations/floors[/{id}]`). Then the round-trip is
pull → edit → validate → publish:

```
pegasus-workflows integration-config pull demo_partner -C ./demo_partner
# …edit mapping.json / rules.json / inbound.json…
pegasus-workflows integration-config validate demo_partner -C ./demo_partner
pegasus-workflows integration-config publish demo_partner -C ./demo_partner
```

#### Forking a platform default — and re-syncing it later

`fork` copies the platform **GLOBAL** config into your tenant as your own overlay
(stamped with fork provenance, after re-running the gate against the current
floor). That overlay then **shadows** GLOBAL for your tenant.

```
pegasus-workflows integration-config fork demo_partner                  # seed my overlay
pegasus-workflows integration-config fork demo_partner --force          # re-sync it (prompts)
```

```python
client.fork_integration_config("demo_partner")               # -> the new TENANT config
client.fork_integration_config("demo_partner", force=True)   # refresh from the current GLOBAL
```

Without `--force` / `force=True` a fork is **one-shot**: if you already own an
overlay you get `409 CONFLICT`, so an overlay forked from an old GLOBAL can never
pick up upstream fixes. `--force` re-seeds it from the **current** GLOBAL as a
**new version** — your previous versions stay in `versions` and remain reachable
via `rollback`, so a refresh you regret is reversible. A stale overlay is easy to
miss (it silently validates against the old contract while GLOBAL is correct), so
re-sync after a platform fix rather than hand-republishing a copy.

Choosing between the two withdrawal/refresh paths:

- **`fork --force`** — you still want your own overlay, just rebased on the latest
  GLOBAL. History preserved; you keep diverging from there.
- **`delete`** (below) — you want _no_ overlay and to re-inherit GLOBAL live from
  then on. The lineage is destroyed.

Either way the gate re-runs against the current floor, so neither can resurrect a
GLOBAL config the contract has outgrown (`422`).

#### Removing an integration

`delete` is the withdrawal path publish/rollback never had — **one verb, scoped by
who calls it**, removing only the config lineage your own tenant owns:

```
pegasus-workflows integration-config delete demo_partner            # prompts first
pegasus-workflows integration-config delete demo_partner --yes --force
```

```python
client.delete_integration_config("demo_partner")               # -> {integrationId, visibility, deleted}
client.delete_integration_config("demo_partner", force=True)   # despite dependent tenant overlays
```

- **Platform tenant → the GLOBAL config.** Retires a placeholder or renamed id
  (e.g. `demo_partner` after a rename to `weichert`) so it stops resolving, drops
  out of `list_integrations()`, and can no longer be forked by tenants.
- **Any other tenant → its own TENANT overlay.** Afterwards
  `get_integration_config(id)` returns the platform GLOBAL again — the supported
  way to drop a stale overlay and re-inherit upstream, rather than
  hand-republishing a copy that never tracks GLOBAL.

It is **irreversible**: the whole version lineage is hard-deleted, so `versions`
comes back empty and `rollback` cannot undo it; a later publish starts again at
`v1`. Deleting a GLOBAL that other tenants still overlay returns `409
DEPENDENTS_EXIST` unless you pass `--force` / `force=True` — which acknowledges
them but never touches their rows. An id that also has a **built-in** definition
in platform code keeps resolving to that code baseline (it stays listed with
`published: false`); a config-only id disappears entirely — `get`,
`map_from_external` and `fork` all 404.

An AI coding agent can do all of this **without platform source**: the MCP
resources `pegasus://reference/integration-config` (the full guide),
`pegasus://reference/floors` (live floors), and `pegasus://reference/openapi` (the
API's OpenAPI 3.1 spec, also at `GET /openapi.json` / Swagger UI `/docs`) carry the
complete contract.

`publish`/`rollback`/`delete` require the token's tenant to be the **platform
tenant** to write GLOBAL (visibility is derived server-side) and to carry the
`PublishIntegrationConfig` action; they are gated by the server's
`INTEGRATION_CONFIG_PUBLISH_ENABLED` switch. `fork` takes the same action and
switch but is the mirror image on tenancy — it writes a TENANT overlay, so the
platform tenant (which owns GLOBAL already) cannot call it. `validate` and `pull`
are read-level and never gated.

## Local Temporal

`pegasus-workflows test` needs a Temporal server. The repo root ships
`docker-compose.temporal.yml` (Temporal server + Temporal UI on `7233` / `8080`)
purely as a local-dev aid — no production connection. `test` runs
`docker compose -f docker-compose.temporal.yml up -d` automatically if Temporal
is not already reachable on `127.0.0.1:7233`. To start it by hand:

```
docker compose -f docker-compose.temporal.yml up -d
```

The Temporal Web UI is then at <http://localhost:8080>.

## Testing activities offline

`pegasus-workflows test` runs a workflow end-to-end against a local Temporal, but
it injects **no** runtime client — so every activity that builds one via
`PegasusClient.from_runtime()` gets nothing and falls back to a hand-written stub,
exercising control flow only. To run an activity's **real** body — a real mapping,
a real read — without any Docker, network, or side effect, use the
`pegasus_workflows.testing` harness:

```python
from pegasus_workflows.testing import fake_client, run_activity
from my_workflow.workflow import fetch_order, send_order_to_partner

# Reads are served from fixtures; keyed reads (get_order, get_secret,
# map_to_external, …) take a {key: value} map, list/whole-value reads take the
# value as-is.
client = fake_client(reads={"get_order": {"S-123": {"orderNumber": "S-123"}}})

# The activity's REAL body runs (inside Temporal's ActivityEnvironment) — not a
# stub — with the fake injected in place of PegasusClient.from_runtime().
order = run_activity(fetch_order, "S-123", client=client)
assert order["orderNumber"] == "S-123"
assert client.captured == []                 # a read — nothing was sent

# Mutations are captured, never performed. Each entry records its Cedar capability.
run_activity(send_order_to_partner, order, client=client)
assert client.captured[0]["capability"] == "SendSms"
```

This retires the `if client is None: return {"stub": True}` pattern: the stub
logic lives in the harness, not in shipped source, so a test exercises the same
code that runs in production. Reads (`get_order`, `map_to_external`,
`get_config`/`get_secret`, `list_*`, …) are benign and served from fixtures;
mutations (`send_sms`, `emit_event`, `close_task`, `put_projection`, …) are
captured to `client.captured` — the same read-vs-mutation split the platform's
Cedar `required_actions` gating enforces, and the same `is_dry_run` /
`record_side_effect` surface the server-side dry-run mode exposes.

**Rehearse the real thing with `--dry-run`.** The offline harness runs one
activity; to rehearse a whole workflow end-to-end on the platform — real reads,
mutations captured, nothing performed — start it in dry-run mode:

```
pegasus-workflows run send_order_to_partner --dry-run --input '{"saleId":"S-123"}'
```

The workflow runs on the tenant runner exactly as a live run would, but the
runtime injects a dry-run client (`client.is_dry_run` is `True`), so reads hit
the live API while every mutation is captured instead of performed. The result
carries the per-activity trace and the capture log of would-be side effects.
Only tenant-runner workflows support it (a curated workflow returns 422
`DRY_RUN_UNSUPPORTED`).

## Using the SDK with an AI coding agent

The SDK ships a built-in [MCP](https://modelcontextprotocol.io/) server that
gives any MCP-compatible AI coding agent (Claude Code, Cursor, Windsurf, …)
structured access to SDK rules and safe tooling — without the agent having to
read or guess from source files.

### Install

The MCP server ships in the base package — no extra needed:

```
pip install pegasus-workflows-sdk
```

(A legacy `[mcp]` extra still resolves as a no-op alias, so older
`pip install 'pegasus-workflows-sdk[mcp]'` commands keep working.)

### Configure your agent

For Claude Code, `pegasus-workflows setup` writes the stanza for you (a `pegasus`
entry in project `./.mcp.json`) — see [First-run setup](#first-run-setup). Use
`pegasus-workflows setup --print-mcp-config` to emit the stanza for any other
host, or wire it by hand below.

#### Claude Code (`~/.claude/settings.json` or project `.claude/settings.json`)

```json
{
  "mcpServers": {
    "pegasus-workflows": {
      "command": "pegasus-workflows",
      "args": ["mcp"]
    }
  }
}
```

#### Cursor / Windsurf (`.cursor/mcp.json` or `.windsurf/mcp.json`)

```json
{
  "mcpServers": {
    "pegasus-workflows": {
      "command": "pegasus-workflows",
      "args": ["mcp"]
    }
  }
}
```

Once configured, the agent can call the resources and tools below without
any additional setup.

### Resources (read-only context)

| URI                              | Description                                                                                                           |
| -------------------------------- | --------------------------------------------------------------------------------------------------------------------- |
| `pegasus://guide/authoring`      | Authoring guide: import surface, determinism rule, input contract pointer.                                            |
| `pegasus://guide/input-contract` | The three `run()` input shapes (trigger-fired, manual run, CLI test) + a worked resolver example.                     |
| `pegasus://guide/secrets-config` | How to publish and use per-tenant workflow secrets & configuration (manifest actions, CLI/SDK publish, runtime read). |
| `pegasus://reference/manifest`   | Manifest fields and constraints generated from `manifest.py` constants — stays in sync automatically.                 |
| `pegasus://reference/api`        | `PegasusClient` method signatures and docstrings generated by introspection — stays in sync automatically.            |

### Tools (safe actions — no network writes)

| Tool                                                                                   | Description                                                                                            |
| -------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `scaffold_workflow(name, dest)`                                                        | Scaffold a new workflow project at `dest/name`. Wraps `pegasus-workflows init`.                        |
| `validate_manifest(path_or_toml)`                                                      | Validate a manifest file path or raw TOML text. Returns structured errors or the parsed manifest.      |
| `package_project(project_dir)`                                                         | Package declared workflows into `dist/`. Returns `{name, version, zip_path, size_bytes}` per workflow. |
| `validate_integration_config(integration_id, mapping, rules, corpus, base_url, token)` | Dry-run the integration config publish gate. No state change.                                          |
| `list_deployments(project_dir)`                                                        | Read a project's `deployments.toml` ledger (no network, no write).                                     |
| `list_profiles()`                                                                      | List credential profile names + api_root. **Never** returns `api_key`.                                 |

Network-mutating operations (`push`, `publish_integration_config`, `run`) are
intentionally **not exposed** — keep those human-gated via the CLI. Secrets never
cross the MCP boundary: `list_profiles` exposes profile names and api_roots only.

### Smoke test (verify the server starts)

```bash
pegasus-workflows mcp --help   # should print the mcp command help
```

Without the extra installed, the command exits non-zero with an install hint.

## Keeping the SDK in sync (maintainers)

The SDK is the **external product boundary** for integration/workflow authors —
they and their AI agents must be able to use the **full** platform functionality
without platform source, via this README, the authoring repo's `CLAUDE.md`, the MCP
`pegasus://reference/*` resources, CLI `--help`, and the API's OpenAPI spec
(`/openapi.json`). The `CLAUDE.md` above is the one in the **workflow-authoring
consumer repo** — a separate repo from `packages/workflows-sdk-python`, so looking
for it inside the SDK or platform source will come up empty. So whenever an
integrations/workflows platform feature is added or changed (a new route, floor,
rule operator, config field, or capability):

1. Expose it in the SDK (a `PegasusClient` method / CLI command / config-file surface).
2. Update its **discovery surfaces in turn** — this README, the authoring repo's
   `CLAUDE.md`, the MCP resources, CLI `--help`, and the OpenAPI spec — preferring
   **live introspection** (e.g. the floor endpoints, and `pegasus://reference/api`,
   which is generated by introspecting `PegasusClient` at serve time) over static
   docs where the contract is code.

A capability that exists in the API but isn't reachable + discoverable through
these is a gap, not a feature.

## Release

The SDK is published to PyPI by `.github/workflows/release-sdk-python.yml` on
`sdk-python-v*` tags via PyPI **trusted publishing** (OIDC — no API token).

To cut a release:

1. Bump `version` in `pyproject.toml` and commit it on `main`.
2. Tag the release commit and push the tag, e.g.
   `git tag sdk-python-v0.1.0 && git push origin sdk-python-v0.1.0`.

The workflow then lints, audits, tests, builds, and uploads the sdist + wheel.

**One-time setup (before the first release):** a PyPI project owner must add a
pending publisher at `pegasus-workflows-sdk` → Publishing → owner `DolasDev`,
repo `pegasus`, workflow `release-sdk-python.yml`, environment `pypi`. Until
that exists the `publish` job fails at the upload step, and tenants must use the
[git install](#interim--unreleased-install-git) above.
