pipegate 0.1.0

Async workflow orchestration with human gates, durable resume, and optional AWS/Slack transports.

pipegate

Workflow orchestration for production AI — human gates and crash-safe resume semantics as first-class primitives.

Pipegate is a workflow framework for multi-stage, async systems where steps can call models, tools, or plain code, with human approval gates and durable resume after restarts or callbacks. It sits alongside LLM and agent libraries: you bring your stack (SDKs, LangChain-style chains, direct API calls). Pipegate owns routing, persisted job state, and pause/resume — not prompt design or autonomous agent loops.

Stages are Tools — the natural place to plug in AI (generation, classification, embedding, eval). The YAML-driven Pipeline is deliberately boring infrastructure today; the roadmap includes optional workflow-level AI config (shared model defaults, tracing hooks, prompt metadata) so many tools can stay declarative without retyping the same wiring.

Stripe just shipped approval gates as the core trust primitive for agentic commerce. Most teams building production AI workflows need the same pattern. Pipegate makes it a YAML config.

pip install pipegate

from pipegate import Pipeline

Pipeline.from_yaml("pipeline.yaml").run()

---

The problem

Most teams building production AI workflows write the same boilerplate:

# What everyone writes today
while True:
    msg = queue.receive()
    if msg.stage == "analyze":
        run_analyze()
    elif msg.stage == "plan":
        run_plan()
    elif msg.stage == "approve":
        send_slack()        # how do you pause here?
        wait_for_human()    # how do you resume?
    elif msg.stage == "execute":
        run_execute()

Switch cases for stages. Custom state threading. Approval gates bolted on as an afterthought. No crash recovery. Every project reinvents the same plumbing.

pipegate eliminates all of it.

---

How it works

Trigger (HTTP / SQS / cron)
    ↓
Stage 1 → your Tool runs → output saved to state
    ↓
Stage 2 → your Tool runs → reads Stage 1 output
    ↓
Gate   → routes to human → pipeline PAUSES ⏸
    ↓  (human approves)
Stage 3 → your Tool runs → reads all previous state
    ↓
Done ✓

You define this in YAML. pipegate owns the consumer loop, stage dispatch, state management, and gate lifecycle. You write the tools.

---

Quickstart

Zero infrastructure — runs locally out of the box

# pipeline.yaml
pipeline:
  name: my-first-pipeline

  consumer:
    type: http          # built-in FastAPI server, no queue needed
    port: 8000

  state_store:
    type: memory        # in-memory, no database needed

  stages:
    - name: analyze
      type: tool
      tool: my_analyzer

    - name: human_review
      type: gate
      transport: webhook
      url: "https://yourapp.com/approve"
      timeout: 8h
      on_approve: execute
      on_reject: abort

    - name: execute
      type: tool
      tool: my_executor

# main.py
from pipegate import Pipeline

class MyAnalyzer:
    name = "my_analyzer"

    async def run(self, job):
        anomalies = await fetch_and_analyze()
        return {"anomalies": anomalies}


class MyExecutor:
    name = "my_executor"

    async def run(self, job):
        anomalies = job.state["analyze"]["anomalies"]
        result = await apply_fix(anomalies)
        return {"fix_applied": result}


Pipeline.from_yaml("pipeline.yaml") \
    .register(MyAnalyzer()) \
    .register(MyExecutor()) \
    .run()

# Start the pipeline — HTTP server ready immediately
python main.py

# Trigger a run
curl -X POST http://localhost:8000/pipeline/run \
  -H "Content-Type: application/json" \
  -d '{"job_id": "job-001"}'

# Deliver a human verdict (approve / reject / modify)
curl -X POST http://localhost:8000/pipeline/verdict/job-001 \
  -H "Content-Type: application/json" \
  -d '{"action": "approve"}'

No AWS account. No Slack app. No queue. Running in 5 minutes.

---

Write your tools

A tool is a Python class with one method. That's the entire contract.

from pipegate import PipelineJob

class AnomalyAnalyzer:
    name = "my_analyzer"

    async def run(self, job: PipelineJob) -> dict:
        # job.state has every output from previous stages
        anomalies = await fetch_and_analyze()
        return {"anomalies": anomalies}   # saved to state automatically


class FixExecutor:
    name = "my_executor"

    async def run(self, job: PipelineJob) -> dict:
        # read previous stage output from state — keyed by stage name
        anomalies = job.state["my_analyzer"]["anomalies"]
        result = await apply_fix(anomalies)
        return {"fix_applied": result}

Tools can do anything — LLM calls, API calls, database writes, shell scripts. pipegate does not care.

Stopping the pipeline from inside a tool

Return None to stop immediately (pipeline marked failed).

Return {"_stop": True, "_status": "SKIPPED", "reason": "..."} to stop with recorded state — the reason is saved to job.state so you can inspect it later.

async def run(self, job: PipelineJob) -> dict | None:
    if not is_eligible(job):
        return {
            "_stop": True,
            "_status": "SKIPPED",
            "reason": "ticket did not meet eligibility criteria",
        }
    result = await do_work(job)
    return {"output": result}

---

Core concepts

Trigger

What starts a pipeline run. Core ships HTTP (built-in FastAPI server). Bring your own via the MessageConsumer protocol, or install a contrib package.

Stage

An ordered step. Each stage calls one registered Tool, saves its output to job.state[stage_name], and passes control to the next stage.

Tool

A Python class you write. One method: async def run(self, job: PipelineJob) -> dict | None. Return a dict, pipegate saves it to job.state[stage_name]. The entire contract.

class Tool(Protocol):
    name: str
    async def run(self, job: PipelineJob) -> dict | None: ...

Gate

A human checkpoint. The pipeline pauses, routes to a human via your chosen transport, and resumes based on the verdict. No threads held. Fully async. Crash-safe — the job record persists in the state store while the pipeline is parked.

class GateTransport(Protocol):
    name: str
    async def send(self, job: PipelineJob, config: GateConfig) -> None: ...
    async def on_verdict(self, job_id: str, action: str, payload: dict) -> None: ...

Verdicts: "approve" advances to on_approve stage. "reject" aborts. "modify" re-enters on_modify stage with feedback in job.state["_message"]["feedback"].

State

Every tool's output is saved to job.state[stage_name] — a dict that travels through all stages. In-memory by default. Swap to DynamoDB or Redis via one line of YAML. A crashed pipeline re-reads state from the store and resumes from the last completed stage.

Sentinel (coming soon)

Declarative policy engine that decides when a gate fires. Gates skip automatically when conditions are met.

sentinel:
  rules:
    - match: execute_payment
      skip_if:
        amount_lt: 50
        merchant_in: approved_merchants

This is how Stripe's agent wallet works — full approval for unknown merchants, automatic for trusted ones within limits. pipegate makes it declarative.

---

Going to production

AWS (SQS + DynamoDB)

pip install pipegate[aws]

pipeline:
  name: production-pipeline

  consumer:
    type: sqs
    queue_url: "${QUEUE_URL}"

  state_store:
    type: dynamodb
    table: pipegate_jobs

  stages:
    - name: analyze
      tool: my_analyzer
    - name: human_review
      type: gate
      transport: slack
      channel: "#ops-approvals"
      timeout: 8h
      on_approve: execute
      on_reject: abort
    - name: execute
      tool: my_executor

Slack gate

pip install pipegate[slack]

from pipegate.contrib.slack import SlackGate

gate = SlackGate(
    name="slack",
    slack_client=my_slack_client,
    build_blocks=my_block_builder,   # returns Slack Block Kit JSON for this job
    enqueue=my_enqueue_fn,
    state_store=state_store,
)

Pipeline.from_yaml("pipeline.yaml") \
    .register(MyAnalyzer()) \
    .register(gate) \
    .register(MyExecutor()) \
    .run()

SlackGate is parameterised — build_blocks is injected, not baked in. The Slack message content stays in your application. pipegate owns the suspend/resume lifecycle.

---

Plugin registry

pipegate ships a plugin registry so any package can register consumer, store, or gate types without pipegate knowing about them.

from pipegate.registry import register_consumer_type, register_store_type, register_gate_type

# In your own package's __init__.py:
register_consumer_type("rabbitmq", lambda cfg: RabbitMQConsumer(cfg["url"]))
register_store_type("redis",       lambda cfg: RedisStateStore(cfg["url"]))
register_gate_type("email",        lambda cfg: EmailGate(cfg["smtp_host"]))

Then in YAML:

consumer:
  type: rabbitmq
  url: "${RABBITMQ_URL}"

pipegate does not need to know RabbitMQ exists. The registry lookup happens at runtime. contrib/aws and contrib/slack use the same mechanism — they self-register at import time.

---

State store options

# In-memory (default, zero deps)
state_store:
  type: memory

# DynamoDB (requires pipegate[aws])
state_store:
  type: dynamodb
  table: pipegate_jobs
  region: us-east-1

# Redis (coming soon)
state_store:
  type: redis
  url: "${REDIS_URL}"

Implement PipelineStateStore (three methods: save, load, create_if_missing) to bring your own.

---

Why not LangGraph?

	LangGraph	pipegate
Define pipeline	Python graph code	YAML
Trigger	You wire it	Built-in HTTP; SQS via contrib
Human gates	Framework-coupled interrupt	First-class, transport-agnostic
State store	LangGraph-managed	Pluggable — memory, DynamoDB, Redis
Framework lock-in	LangChain ecosystem	None — bring your own tools
Crash recovery	Checkpoint API	Built-in via state store
Local dev	Needs LangGraph runtime	Zero infra, HTTP trigger out of the box
Gate content	Framework controls message	build_blocks injected — you control content

---

Built with pipegate

PARP — Proactive Application Reliability Pipeline. Monitors production anomalies, generates fix candidates, routes through a Slack approval gate, and opens PRs. Running in production.

pipeline:
  name: parp
  consumer:
    type: sqs
    queue_url: "${PARP_QUEUE_URL}"
  state_store:
    type: dynamodb
    table: pipegate_jobs
  stages:
    - name: triage
      type: trigger       # fans out: one POLL_METRICS → N anomaly jobs
      tool: datadog_scanner
    - name: leadership_gate
      type: gate
      transport: slack
      channel: "#leadership"
      timeout: 72h
      on_approve: classify
      on_reject: abort
    - name: classify
      tool: llm_classifier
    - name: context
      tool: context_builder
    - name: plan
      tool: bedrock_fix_planner
    - name: approval_gate
      type: gate
      transport: slack
      channel: "#eng-approvals"
      timeout: 24h
      on_approve: codegen
      on_reject: abort
      on_modify: plan     # re-enter plan with engineer feedback
    - name: codegen
      tool: llm_code_generator
    - name: create_pr
      tool: github_pr_creator

---

Package structure

pip install pipegate          # core only — zero infra deps
pip install pipegate[aws]     # + SQSConsumer, DynamoDBStateStore
pip install pipegate[slack]   # + SlackGate
pip install pipegate[all]     # everything

Core dependencies: pyyaml, fastapi, uvicorn. No cloud SDK. No Slack SDK. The HTTP trigger and in-memory state store work with zero extras installed.

---

Roadmap

• HTTP trigger (built-in)
• YAML pipeline definition
• In-memory state store
• Stage dispatcher — no switch cases, no boilerplate
• _stop / _status pipeline control from tools
• SQS consumer (pipegate[aws])
• DynamoDB state store with GSI support (pipegate[aws])
• Slack gate transport (pipegate[slack])
• Plugin registry — register_consumer_type, register_store_type, register_gate_type
• Cron trigger
• Redis state store
• Webhook gate transport
• Sentinel policy engine
• pipegate[openai] tool plugin
• pipegate[bedrock] tool plugin
• Visual pipeline inspector

---

Contributing

pipegate is early. If you are building agentic pipelines and hit a wall, open an issue. PRs welcome.

git clone https://github.com/yourusername/pipegate
cd pipegate
pip install -e ".[dev]"
pytest

---

License

MIT