postgrip-agent 0.11.0

Python SDK for PostGrip Agent workflows

PostGrip Agent Python SDK

Python SDK for defining, submitting, and executing PostGrip workflows. In production, SDK workflow runtimes are supervised by an existing PostGrip agent: the host agent launches the runtime, injects delegated credentials, and keeps generic operational tasks separate from workflow/activity task polling. Client-side SDK code submits workflow.runtime tasks to an existing agent pool; it does not enroll or spawn standalone PostGrip agents. 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.

Current release: 0.11.0

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 / 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
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:
    # This process is launched by a PostGrip host agent from a workflow.runtime
    # task. The host injects POSTGRIP_AGENT_ID and delegated runtime credentials.
    client = await Client.connect()
    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",
            ui={
                "displayName": "Say hello to PostGrip",
                "description": "Shown on the PostGrip Agents activity tab.",
                "details": {"sdk": "python"},
                "tags": ["demo"],
            },
        )
    )
    print(result)


asyncio.run(main())

ui is SDK-owned console metadata. It is persisted inside workflow memo as postgrip.ui, so the Agents activity tab can show a friendly label, description, details, and tags while memo remains available for your own data.

Submit that runtime to an existing agent pool from your client process:

import asyncio
import os

from postgrip_agent import Client


async def submit_runtime() -> None:
    client = await Client.connect(
        # Agent token from Settings > Organization > Agent tokens.
        headers={"Authorization": f"Bearer {os.environ['POSTGRIP_AGENT_TOKEN']}"},
    )
    client.task.workflow_runtime(
        queue="default",
        command="python",
        args=["-m", "myapp.workflow_runtime"],
        runtime_queue="default",
        env={"POSTGRIP_EXAMPLE_RUN_LABEL": "PostGrip"},
    )


asyncio.run(submit_runtime())

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(...); activities can also attach task output with activity.stdout(...) and activity.stderr(...). 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. For SDK applications, the documented client-side submission path is client.task.workflow_runtime(...); workflow and activity tasks are then coordinated by the managed runtime launched on the host agent.

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