postgrip-agent 0.1.0

Python SDK for PostGrip Agent workflows

PostGrip Agent Python SDK

Companion of agent-sdk-go and agent-sdk-typescript. Wire shapes are tracked in agent-sdk-protocol.

Docs: postgrip-io.github.io/agent-sdk-python — quick start, workflow runtime, API guide. PyPI: pypi.org/project/postgrip-agent.

Install from PyPI after publishing:

pip install postgrip-agent

For local development from a clone of this repository:

pip install -e .
PYTHONPATH=src python -m unittest discover -s test

Layout

src/postgrip_agent/   # Python package — Connection / Client / Agent / workflow runtime
test/                 # unittest-style tests
doc/                  # reserved for longer-form prose docs
.github/workflows/    # CI: build wheel + run tests on 3.11 / 3.12 / 3.13

The package exposes a Temporal-style Python API:

import asyncio
import os
from datetime import timedelta

from postgrip_agent import Client, Agent, activity, workflow


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


@workflow.defn(name="SayHelloWorkflow")
class SayHelloWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            greet,
            name,
            schedule_to_close_timeout=timedelta(seconds=10),
        )


async def main() -> None:
    client = await Client.connect(
        "http://127.0.0.1:4100",
        headers={"Authorization": f"Bearer {os.environ['POSTGRIP_AGENT_AUTH_TOKEN']}"},
    )
    agent = Agent(
        client,
        task_queue="default",
        workflows=[SayHelloWorkflow],
        activities=[greet],
    )
    result = await agent.run_until(
        client.execute_workflow(
            SayHelloWorkflow,
            "PostGrip",
            id="say-hello",
            task_queue="default",
        )
    )
    print(result)


asyncio.run(main())

This SDK targets the PostGrip Agent runtime API, not a Temporal server. It follows the familiar Temporal Python shape for client, agent, workflow, and activity code while using PostGrip Agent task queues and workflow history underneath.

Implemented workflow APIs include durable activity scheduling/replay, durable timers via workflow.sleep(), query/signal/update handler replay, child workflow scheduling/replay, continue-as-new, cancellation scopes, and command-order determinism checks for activities, timers, and children.

The Python agent supports bounded concurrent task execution with max_concurrent_tasks defaulting to 4, graceful shutdown draining for in-flight tasks with an optional timeout, and automatic lease renewal for workflow, query, update, and activity tasks. Activities and workflows can emit ordered milestones with activity.milestone("step name", index=1, total=10) or workflow.milestone(...); clients can stream those task events with handle.watch_events() or client.task.watch_events(task_id). Workflow execution also performs sandbox checks that reject common nondeterministic APIs such as time.time(), time.sleep(), asyncio.sleep(), random.*(), and uuid.uuid4(); use workflow.now(), workflow.sleep(), explicit IDs, or activities for those operations.

Public protocol types are available from postgrip_agent.types and are re-exported from postgrip_agent, including Task, TaskEvent, WorkflowExecution, WorkflowHistoryEvent, RetryPolicy, schedule request/response types, and workflow payload definitions. The package includes py.typed so type checkers can consume those annotations.

Lower-level task API:

client.task.shell_exec(command="echo", args=["hello from agent"])

# Polyglot: the Go agent runs the command inside a per-task container
# (proxied through the worker stack's docker socket proxy). The agent
# process must have DOCKER_HOST set; the container runs with --rm
# --network=none and no host mounts. The env-key allowlist rejects
# DOCKER_*, POSTGRIP_*, and host loader/interpreter prefixes.
client.task.container_exec(
    image="node:22-alpine",
    command="node",
    args=["-e", "console.log('hi from node')"],
    pull_policy="missing",
    timeout_seconds=60,
)

Package validation:

python -m pip wheel --no-deps postgrip-agent/python -w /tmp/postgrip-agent-wheel
PYTHONPATH=postgrip-agent/python python -m unittest discover -s postgrip-agent/python/tests