tryagent 0.0.2

Python SDK for creating and managing TryAgent escalations.

tryagent

Python SDK for creating and managing TryAgent escalations from agents, workflows, and reviewer tooling.

Full documentation: https://tryagentai.mintlify.app/quickstart

Install

python3 -m pip install tryagent

From a checkout, install it locally with python3 -m pip install -e packages/python-sdk.

Create a client

import os

from tryagent import TryAgent

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

Override base_url for local development or tests:

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

Send an escalation

Request payloads use the TryAgent API's camelCase fields, even from Python.

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"])

Resume a workflow

Always pass the raw request body to tryagent.webhooks.construct_event. Parsing and re-serializing JSON changes the signed bytes.

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

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."})

Handle API errors

Non-2xx API responses and network failures raise ApiError. Network failures use status == 0; API responses use the HTTP status code. body contains the parsed response body when available, and request_id is read from the x-request-id response header when present.

from tryagent import ApiError

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