Python

Create and manage TryAgent escalations from Python agents and workflows.

Python

Use the Python SDK from agents, workflow workers, and reviewer tools that need to create or manage TryAgent escalations.

Install

The Python SDK is available in this repository and is not published to PyPI yet. From a checkout, install it locally:

python3 -m pip install -e packages/python-sdk

After the first PyPI release, install it with:

python3 -m pip install tryagent

Create a client

import os

from tryagent import TryAgent

tryagent = TryAgent(api_key=os.environ["TRYAGENT_API_KEY"])

Client options are Pythonic snake_case:

tryagent = TryAgent(
    api_key=os.environ["TRYAGENT_API_KEY"],
    base_url="http://localhost:4000",
)

Provide exactly one of api_key, token, or get_token.

Send an escalation

Payload fields are the TryAgent API fields, so keep them camelCase inside the request body.

import os

from tryagent import TryAgent

tryagent = TryAgent(api_key=os.environ["TRYAGENT_API_KEY"])

escalation = tryagent.escalate(
    "orders.auth_doc",
    {
        "agentId": "order-agent",
        "runId": "run_4821",
        "subject": {
            "type": "order",
            "id": "ord_4821",
            "label": "Order #4821",
        },
        "question": "The authorization document is missing a signature date. Continue?",
        "evidence": [
            "Candidate name is present.",
            "Employer is present.",
            "Signature date is blank.",
        ],
        "choices": [
            {"id": "manual_review", "label": "Send to manual review"},
            {"id": "continue", "label": "Continue anyway"},
        ],
        "resume": {
            "mode": "webhook",
            "url": "https://api.example.com/tryagent/resume",
            "secret": os.environ["TRYAGENT_WEBHOOK_SECRET"],
        },
    },
)

print(escalation["id"], escalation["status"])

Use tryagent.escalations.create({...}) when you already have a complete input object that includes policy.

Manage escalations

open_escalations = tryagent.escalations.list(status="open")

escalation = tryagent.escalations.get("esc_123")

tryagent.escalations.acknowledge("esc_123")

tryagent.escalations.decide(
    "esc_123",
    {
        "choice": "continue",
        "reason": "Signature verified out of band.",
        "response": {"approvedLimit": 5000, "riskLevel": "low"},
    },
)

tryagent.escalations.cancel("esc_456", {"reason": "Order withdrawn."})

Verify resume callbacks

When a reviewer decides, TryAgent sends a signed callback to your resume.url. Pass the raw body bytes or string to construct_event; do not parse JSON before verification.

import os

from tryagent import TryAgent, WebhookSignatureError

tryagent = TryAgent(api_key=os.environ["TRYAGENT_API_KEY"])


def handle_resume(raw_body: bytes, signature: str | None) -> tuple[dict[str, bool], int]:
    try:
        event = tryagent.webhooks.construct_event(
            payload=raw_body,
            signature=signature,
            secret=os.environ["TRYAGENT_WEBHOOK_SECRET"],
        )
    except WebhookSignatureError:
        return {"ok": False}, 401

    resume_workflow(
        event["runId"],
        {
            "escalationId": event["escalationId"],
            "choice": event.get("choice"),
            "answeredBy": event.get("answeredBy"),
            "answeredAt": event.get("answeredAt"),
        },
    )
    return {"ok": True}, 200

You can also use the standalone helpers:

from tryagent import construct_webhook_event, verify_webhook_signature

ApiError

The SDK raises ApiError for non-2xx API responses and network failures.

from tryagent import ApiError

try:
    tryagent.escalate("orders.auth_doc", input)
except ApiError as error:
    print(error.status, error.request_id, error.body)
    raise

error.status is 0 for network failures. For API responses, error.body contains the parsed response when available and error.request_id comes from the response headers when present.