bridge-pipelines

name

bridge-pipelines

description

Build and configure Bridge pipelines using the Bridge SDK. Use when the task involves creating pipeline steps, defining step dependencies (DAGs), integrating agents into pipelines, configuring credential bindings, defining evals for quality measurement, setting up pyproject.toml for Bridge, or working with the Bridge CLI (bridge check, bridge config get-dsl, bridge run, bridge eval run). Also use when connecting pipelines to the Poolside web interface via repository indexing and build execution.

Bridge Pipelines

Bridge is a workflow orchestration framework for defining and executing multi-step data processing pipelines with automatic dependency management. Pipelines are defined in Python using the Bridge SDK, stored in a git repository, and executed via the Poolside web interface.

End-to-End Flow

1. Define pipeline steps in Python using the Bridge SDK
2. Configure pyproject.toml with [tool.bridge] section
3. Push code to a git repository
4. Register the repository in the Poolside web interface (Bridge UI)
5. Index a commit — the system runs SDK analysis to discover pipelines and steps
6. Create a build to execute the pipeline DAG

Quick Start

Project Setup

uv init my_project
cd my_project
uv add bridge-sdk@git+https://github.com/poolsideai/bridge-sdk.git

Configure pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.bridge]
modules = ["my_project.steps"]

The modules list tells Bridge which Python modules to import when discovering steps.

Define a Pipeline

# my_project/steps.py
from typing import Annotated
from pydantic import BaseModel
from bridge_sdk import Pipeline, step_result

pipeline = Pipeline(
    name="my_pipeline",
    rid="550e8400-e29b-41d4-a716-446655440000",  # optional stable ID
    description="My processing pipeline",
)

class InputData(BaseModel):
    value: str

class ProcessedData(BaseModel):
    result: str

@pipeline.step
def ingest() -> InputData:
    return InputData(value="raw data")

@pipeline.step
def process(data: Annotated[InputData, step_result(ingest)]) -> ProcessedData:
    return ProcessedData(result=f"processed: {data.value}")

Validate and Test

uv run bridge check          # Validate project setup
uv run bridge config get-dsl # Generate DSL (shows discovered steps/pipelines)
uv run bridge run --step ingest --input '{}' --results '{}'

Core Concepts

Pipeline

A named container grouping related steps. Auto-registers on instantiation.

pipeline = Pipeline(
    name="my_pipeline",
    rid="uuid-here",       # optional — stable identity across renames
    description="...",
)

Steps

Steps are the execution units. Use @pipeline.step to bind a step to a pipeline.

@pipeline.step(
    name="custom_name",                   # override function name
    rid="uuid-here",                      # stable resource ID
    description="What this step does",
    setup_script="scripts/setup.sh",      # runs before step
    post_execution_script="scripts/cleanup.sh",  # runs after step
    metadata={"type": "agent"},           # arbitrary metadata
    credential_bindings={                 # credential UUID → env var
        "cred-uuid": "API_KEY",
    },
)
def my_step(input_data: InputModel) -> OutputModel:
    ...

All parameters and return types must be JSON-serializable (Pydantic models, primitives, collections, dataclasses, enums, Optional, Union).

The decorator supports multiple invocation styles:

@pipeline.step            # no parentheses
@pipeline.step()          # empty parentheses
@pipeline.step(name="x")  # with arguments

Step Dependencies (DAG)

Declare dependencies using step_result annotations. The DAG is automatically inferred.

from typing import Annotated
from bridge_sdk import step_result

@pipeline.step
def upstream() -> DataModel:
    return DataModel(...)

@pipeline.step
def downstream(
    data: Annotated[DataModel, step_result(upstream)],
    other: Annotated[OtherModel, step_result(another_step)],
) -> ResultModel:
    # data is automatically populated from upstream's output
    return ResultModel(...)

A step can depend on multiple upstream steps. Steps with no dependencies run first. The orchestrator executes steps in parallel when their dependencies are satisfied.

Credential Bindings

Inject credentials as environment variables at runtime:

@pipeline.step(
    credential_bindings={
        "a1b2c3d4-5678-90ab-cdef-1234567890ab": "MY_API_KEY",
        "f0e1d2c3-b4a5-6789-0abc-def123456789": "DB_PASSWORD",
    }
)
def secure_step() -> str:
    import os
    api_key = os.environ["MY_API_KEY"]
    return "done"

Keys are credential UUIDs registered in the Bridge UI. Values are the env var names to inject.

WebhookPipelineActions

Pipelines can be triggered by external webhook events. WebhookPipelineAction endpoints (signature verification, secrets, idempotency) are configured in Console. The SDK declares actions that reference an endpoint by name and define filtering/transformation logic via CEL.

from bridge_sdk import Pipeline, WebhookPipelineAction

pipeline = Pipeline(
    name="on_issue_update",
    webhooks=[
        # branch determines where this webhook is indexed from and which
        # version of the pipeline code runs when it fires. The webhook won't
        # exist until this branch is indexed.
        WebhookPipelineAction(
            name="linear-issues",
            branch="main",
            on='payload.type == "Issue" && payload.action == "update"',
            transform='{"triage_step": {"issue_id": payload.data.id, "title": payload.data.title}}',
            webhook_endpoint="linear_issues",
        ),
        WebhookPipelineAction(
            name="github-push",
            branch="production",
            on='payload.ref == "refs/heads/main"',
            transform='{"index_step": {"repo": payload.repository.full_name, "commit_sha": payload.head_commit.id}}',
            webhook_endpoint="github_pushes",
        ),
    ],
)

WebhookPipelineAction fields:

Field	Type	Required	Description
name	str	Yes	Unique name within the pipeline + branch
branch	str	Yes	The git branch this webhook is indexed from and whose pipeline code runs when it fires
on	str	Yes	CEL expression returning bool — action fires only when true
transform	str	Yes	CEL expression returning map(string, map(string, dyn)) — step name to input map
webhook_endpoint	str	Yes	Name of the webhook endpoint configured in Console

CEL expressions receive payload (the parsed JSON body) and headers (HTTP headers as map(string, string)).

Example files: examples/webhook_example.py, examples/webhook_generic_example.py.

Agent Integration

Steps can launch AI agents via the Bridge sidecar gRPC service. See references/agents.md for the full agent integration guide.

from bridge_sdk.bridge_sidecar_client import BridgeSidecarClient

@pipeline.step(metadata={"type": "agent"})
def agent_step() -> dict:
    output_file = "/tmp/agent_output.json"
    with BridgeSidecarClient() as client:
        _, session_id, _ = client.start_agent(
            prompt=f"Analyze data and write results to {output_file}",
            agent_name="Malibu",
        )
    with open(output_file) as f:
        return json.load(f)

Important: start_agent() returns (agent_name, session_id, exit_result). The exit_result is just a status message — not the agent's output. Agents must write results to a file that the step reads.

Evals

Evals measure the quality of step and pipeline outputs. Define evals with @bridge_eval, then attach them via eval_bindings on step/pipeline definitions.

from typing import TypedDict, Any
from bridge_sdk import bridge_eval, EvalResult, StepEvalContext, step, on_branch

class QualityMetrics(TypedDict):
    accuracy: float

@bridge_eval
def quality_check(ctx: StepEvalContext[Any, Any]) -> EvalResult[QualityMetrics]:
    return EvalResult(metrics={"accuracy": 1.0})

@step(eval_bindings=[(quality_check, on_branch("main"))])
def my_step(value: str) -> str:
    return value

See references/evals.md for the full evals reference including conditions, pipeline evals, and metadata types.

CLI Reference

Command	Purpose
bridge check	Validate pyproject.toml and discover steps
bridge config get-dsl	Generate JSON DSL of all pipelines, steps, and evals
bridge run --step NAME --input JSON --results JSON	Execute a single step locally
bridge eval run --eval NAME --context JSON	Execute a single eval locally

See references/cli.md for full CLI details.

Web Interface Integration

See references/web-integration.md for how pipelines connect to the Poolside web interface, including repository management, commit indexing, build execution, and the API.