Metadata-Version: 2.4
Name: pipegate
Version: 0.1.0
Summary: Async workflow orchestration with human gates, durable resume, and optional AWS/Slack transports.
License: MIT
Project-URL: Homepage, https://github.com/YOUR_ORG/pipegate
Project-URL: Documentation, https://github.com/YOUR_ORG/pipegate#readme
Project-URL: Repository, https://github.com/YOUR_ORG/pipegate
Project-URL: Issues, https://github.com/YOUR_ORG/pipegate/issues
Keywords: workflow,async,human-in-the-loop,sqs,llm,ai
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyyaml>=6.0
Requires-Dist: fastapi>=0.110
Requires-Dist: uvicorn>=0.29
Provides-Extra: aws
Requires-Dist: aioboto3>=12.0; extra == "aws"
Provides-Extra: slack
Requires-Dist: slack-sdk>=3.0; extra == "slack"
Provides-Extra: all
Requires-Dist: aioboto3>=12.0; extra == "all"
Requires-Dist: slack-sdk>=3.0; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-mock>=3.12; extra == "dev"
Requires-Dist: fastapi>=0.110; extra == "dev"
Requires-Dist: uvicorn>=0.29; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"
Dynamic: license-file

# 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.

```bash
pip install pipegate
```

```python
from pipegate import Pipeline

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

---

## The problem

Most teams building production AI workflows write the same boilerplate:

```python
# 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

```yaml
# 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
```

```python
# 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()
```

```bash
# 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.

```python
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.

```python
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.

```python
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.

```python
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.

```yaml
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)

```bash
pip install pipegate[aws]
```

```yaml
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

```bash
pip install pipegate[slack]
```

```python
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.

```python
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:

```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

```yaml
# 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.

```yaml
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

- [x] HTTP trigger (built-in)
- [x] YAML pipeline definition
- [x] In-memory state store
- [x] Stage dispatcher — no switch cases, no boilerplate
- [x] `_stop` / `_status` pipeline control from tools
- [x] SQS consumer (`pipegate[aws]`)
- [x] DynamoDB state store with GSI support (`pipegate[aws]`)
- [x] Slack gate transport (`pipegate[slack]`)
- [x] 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.

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

---

## License

MIT
