SDKs · Python

Python SDK

The phrony client for Python — install, run agents and bundles, stream sessions, and register workers.

phrony is the Python client for the Phrony runtime. It targets Python 3.10+, uses grpc.aio for async transport, and maps proto bytes JSON fields through helpers so you work with ordinary Python objects — strings and dicts for agent refs, keyword arguments on common RPCs, and flattened interactive events. All public APIs are async — run them with asyncio.

Install

Install from PyPI, then connect to a runtime. For a guided tool-binding walkthrough, see Add a tool binding.

Package

pip install phrony

Requires Python 3.10+.

Shell

$ pip install phrony

Import paths

phrony and phrony.worker

Use the main package for Phrony, RuntimeClient, AgentRef, BundleRef, InteractiveSession, and Worker. Use phrony.worker when you only need WorkStream and worker types.

Python

# Full SDK
from phrony import (
    AgentRef,
    BundleRef,
    InteractiveEvent,
    InteractiveSession,
    Phrony,
    PhronyAgent,
    PhronyBundle,
    RuntimeClient,
    Worker,
)

# Worker subpackage (includes WorkStream)
from phrony.worker import Worker, WorkStream, ToolError

Connect to the runtime

Every SDK entrypoint dials the runtime over gRPC. Set PHRONY_RUNTIME_ADDR or pass an explicit address. After installing the package, use one of the patterns below. All connection methods are async.

RuntimeClient

await RuntimeClient.connect(...)

Dial the runtime and health service. Options: address, credentials, channel_options. The resolved address is available on client.address.

Python

import asyncio
import os

from phrony import RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect(
        address=os.environ.get("PHRONY_RUNTIME_ADDR", "127.0.0.1:7777"),
    )
    print("Connected to", client.address)
    await client.close()


asyncio.run(main())

await client.close()

Close the gRPC channel when finished.

Python

client = await RuntimeClient.connect()
# ...
await client.close()

Phrony

await Phrony.connect(...)

High-level facade with the same connection options (runtime_addr maps to address).

Python

import asyncio
import os

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect(
        runtime_addr=os.environ.get("PHRONY_RUNTIME_ADDR"),
    )
    # Same options as RuntimeClient: credentials, runtime_addr
    await phrony.close()


asyncio.run(main())

Address resolution

resolve_runtime_addr(address?)

Resolve the gRPC target: explicit argument, then PHRONY_RUNTIME_ADDR, then 127.0.0.1:7777.

Python

from phrony import DEFAULT_RUNTIME_ADDR, resolve_runtime_addr

# Explicit address wins, then PHRONY_RUNTIME_ADDR, then DEFAULT_RUNTIME_ADDR
addr = resolve_runtime_addr()  # e.g. "127.0.0.1:7777"
print(DEFAULT_RUNTIME_ADDR)

dial_runtime(...)

Low-level dial returning channel, runtime stub, health stub, address, and close() without the RuntimeClient wrapper.

Python

import asyncio

from phrony import dial_runtime
from phrony.gen.phrony.runtime.v1 import runtime_pb2


async def main() -> None:
    dial = dial_runtime(address="127.0.0.1:7777")
    try:
        version = await dial.runtime.GetVersion(runtime_pb2.GetVersionRequest())
        print(version)
    finally:
        await dial.close()


asyncio.run(main())

TLS

credentials (optional)

By default the SDK uses an insecure channel. Pass grpc.ssl_channel_credentials when the runtime requires TLS.

Python

import asyncio
import grpc

from phrony import RuntimeClient


async def main() -> None:
    with open("ca.pem", "rb") as ca, open("client-key.pem", "rb") as key, open(
        "client-cert.pem", "rb"
    ) as cert:
        credentials = grpc.ssl_channel_credentials(
            root_certificates=ca.read(),
            private_key=key.read(),
            certificate_chain=cert.read(),
        )

    client = await RuntimeClient.connect(
        address="runtime.example.com:7777",
        credentials=credentials,
    )
    await client.close()


asyncio.run(main())

Health check

client.health()

Standard gRPC health stub on the same address as the runtime.

Python

import asyncio

from phrony import RuntimeClient
from phrony.gen.grpc.health.v1 import health_pb2


async def main() -> None:
    client = await RuntimeClient.connect()
    health = client.health()
    response = await health.Check(health_pb2.HealthCheckRequest(service=""))
    print(response.status)  # SERVING, NOT_SERVING, etc.
    await client.close()


asyncio.run(main())

Cleanup

await Phrony.close() / await RuntimeClient.close()

Python

import asyncio

from phrony import Phrony, RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    await client.close()

    phrony = await Phrony.connect()
    await phrony.close()


asyncio.run(main())

Run agents & bundles

Phrony is the high-level entry point for agent and bundle sessions. Connect first — see Connect to the runtime.

Phrony

Phrony(...)

Create a facade. runtime_addr and optional credentials match RuntimeClient options. The client dials lazily on first use.

Python

from phrony import Phrony

phrony = Phrony(runtime_addr="127.0.0.1:7777")
agent = phrony.agent("default/my-agent")
# Client dials lazily on first run() or runtime_client()

await Phrony.connect(...)

Connect eagerly and return a ready instance.

Python

import asyncio
import os

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect(
        runtime_addr=os.environ.get("PHRONY_RUNTIME_ADDR"),
    )
    # Same options as RuntimeClient: credentials, runtime_addr
    await phrony.close()


asyncio.run(main())

await phrony.close()

Close the underlying gRPC client.

Python

import asyncio

from phrony import Phrony, RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    await client.close()

    phrony = await Phrony.connect()
    await phrony.close()


asyncio.run(main())

phrony.agent("namespace/name")

Return a PhronyAgent. Accepts namespace/name@version. Bare names without a slash raise AgentRefParseError.

Python

phrony = await Phrony.connect()

pinned = phrony.agent("default/my-agent@0.2.0")
latest = phrony.agent("default/my-agent")

phrony.bundle("namespace/name")

Return a PhronyBundle. Accepts namespace/name@version where version is semver or a lock hash (sha256:…). Bare names without a slash raise BundleRefParseError.

Python

phrony = await Phrony.connect()

semver = phrony.bundle("demo/payment-desk@1.0.0")
lock_hash = phrony.bundle("demo/payment-desk@sha256:abc…")
active = phrony.bundle("demo/payment-desk")

await phrony.runtime_client()

Return the lazily connected RuntimeClient.

Python

import asyncio

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()
    client = await phrony.runtime_client()
    agents = await client.list_agents()
    print(agents)


asyncio.run(main())

PhronyAgent

await run(...)

Start a session. wait=True (default) streams until completed; wait=False uses unary RunSession and returns immediately.

Python

import asyncio
import os

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect(
        runtime_addr=os.environ.get("PHRONY_RUNTIME_ADDR", "127.0.0.1:7777"),
    )

    result = await phrony.agent("default/my-agent").run(
        input={"claimId": "CLM-48219"},
        resolved_secrets={"openai": os.environ["OPENAI_API_KEY"]},
    )

    print(result.session_id, result.output, result.stop_reason)
    await phrony.close()


asyncio.run(main())

await run(wait=False)

Fire-and-forget session start.

Python

import asyncio

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()

    result = await phrony.agent("default/my-agent").run(
        input={"claimId": "CLM-48219"},
        wait=False,
    )

    print(result.session_id, result.status)


asyncio.run(main())

await run_interactive(...)

Open RunSessionInteractive without waiting. Consume events() and call close() when done.

Python

import asyncio

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()

    session = await phrony.agent("default/my-agent").run_interactive(
        input={"question": "Summarize claim CLM-48219"},
    )

    async for event in session.events():
        if event["type"] == "text_delta":
            print(event["delta"], end="", flush=True)

    await session.close()
    await phrony.close()


asyncio.run(main())

PhronyBundle

Multi-agent systems published as a Bundle run through the bundle root member. PhronyBundle mirrors PhronyAgent — same run() options and AgentSessionError on failure.

await run(...)

Start a session on the bundle root. wait=True (default) streams until completed; wait=False uses unary RunSession with bundle_ref.

Python

import asyncio
import os

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()

    result = await phrony.bundle("demo/payment-desk").run(
        input={"message": "Pay 500 USD to Acme"},
        resolved_secrets={"stripe": os.environ["STRIPE_API_KEY"]},
    )

    print(result.session_id, result.output, result.stop_reason)
    await phrony.close()


asyncio.run(main())

await run(wait=False)

Fire-and-forget bundle session start.

Python

import asyncio

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()

    result = await phrony.bundle("demo/payment-desk").run(
        input={"message": "Pay 500 USD to Acme"},
        wait=False,
    )

    print(result.session_id, result.status)


asyncio.run(main())

await run_interactive(...)

Open RunSessionInteractive with bundle_ref without waiting.

Python

import asyncio

from phrony import Phrony


async def main() -> None:
    phrony = await Phrony.connect()

    session = await phrony.bundle("demo/payment-desk").run_interactive(
        input={"message": "Pay 500 USD to Acme"},
    )

    async for event in session.events():
        if event["type"] == "text_delta":
            print(event["delta"], end="", flush=True)

    await session.close()
    await phrony.close()


asyncio.run(main())

For streaming event types, see Interactive sessions. On failure with wait=True, run() raises AgentSessionError — see Utilities.

Interactive sessions

InteractiveSession wraps RunSessionInteractive. Open it from RuntimeClient.run_session_interactive() or PhronyAgent.run_interactive() or PhronyBundle.run_interactive(). The first client message must be start or attach.

Methods

await start(...)

New session: agent_ref or bundle_ref as a string (namespace/name), dict, or parsed ref — mutually exclusive. Pass JSON input and optional resolved_secrets as plain dicts.

Python

import asyncio
import os

from phrony import RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    session = client.run_session_interactive()

    await session.start(
        agent_ref="default/my-agent",
        input={"question": "Hello"},
        resolved_secrets={"openai": os.environ["OPENAI_API_KEY"]},
    )

    await session.close()
    await client.close()


asyncio.run(main())

await start({ bundle_ref })

Start an interactive session on a deployed bundle.

Python

import asyncio
import os

from phrony import RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    session = client.run_session_interactive()

    await session.start(
        bundle_ref="demo/payment-desk",
        input={"message": "Pay 500 USD to Acme"},
        resolved_secrets={"stripe": os.environ["STRIPE_API_KEY"]},
    )

    await session.close()
    await client.close()


asyncio.run(main())

await attach(...)

Reconnect to an existing session_id.

Python

await session.attach(session_id="sess_abc123")

await send_user_message(text)

Send a user turn after an awaiting_input event.

Python

# After an awaiting_input event:
await session.send_user_message("Use a shorter summary.")

await decide_tool_approval(...)

Respond to approval_required: approval_id, approved, optional comment and replacement args.

Python

# Inside your events() loop when event["type"] == "approval_required":
await session.decide_tool_approval(
    approval_id=event["approval_id"],
    approved=True,
    comment="Approved by operator",
    args={"city": "Oslo"},
)

async for event in session.events()

Async iterator of event dicts until the stream ends or errors.

Python

async for event in session.events():
    match event["type"]:
        case "session_started":
            print(event["session_id"], event["agent_version_id"])
        case "text_delta":
            print(event["delta"], end="", flush=True)
        case "completed":
            print(event.get("output"))
        case "failed":
            raise RuntimeError(event["message"])

await session.close()

Half-close the client side of the stream.

Python

await session.close()
await client.close()

InteractiveEvent types

Server events are dicts with a type key. Lifecycle and tool events expose flat fields — for example session_started includes session_id, agent_version_id, and history; approval_required includes approval_id, tool, and parsed args.

Lifecycle

session_started, completed, failed, cancelled, stream_end

Streaming

text_delta, awaiting_input

Tools

tool_call, tool_result, approval_required

Tool workers

Workers implement tools declared on agents over the Work stream. See Tool workers in the runtime docs.

Worker

Worker(...)

worker_id is required. Optional runtime_addr, workload_identity, image_digest.

Python

import os

from phrony.worker import Worker

worker = Worker(
    worker_id="weather-worker-1",
    runtime_addr=os.environ.get("PHRONY_RUNTIME_ADDR"),
    workload_identity="k8s://default/weather-worker",
    image_digest="sha256:abc…",
)

@worker.tool(name, *, version, ...)

Primary registration path — decorator sugar over register_tool. Returns the original function unchanged.

Python

@worker.tool("weather.get-forecast", version="1.0.0", max_concurrency=4)
async def get_forecast(args: dict, ctx) -> dict:
    if not args.get("city"):
        raise ToolError("invalid_args", "city is required")
    return {"temp_c": 12, "city": args["city"]}

register_tool(...)

Imperative alternative. Register before connect(). handler receives (args, context) with cancellation via asyncio.

Python

worker.register_tool(
    tool="weather.get-forecast",
    version="1.0.0",
    max_concurrency=4,
    handler=get_forecast,
)

await connect() / await close()

connect() blocks until the stream ends; close() aborts in-flight calls and shuts down.

Python

import asyncio

from phrony.worker import Worker, ToolError


worker = Worker(worker_id="weather-worker-1")


@worker.tool("weather.get-forecast", version="1.0.0")
async def get_forecast(args: dict, ctx) -> dict:
    return {"temp_c": 12, "city": args["city"]}


async def main() -> None:
    await worker.connect()  # blocks until disconnect
    await worker.close()


asyncio.run(main())

ToolError

Raise from a handler to return a structured tool failure to the runtime.

Python

from phrony.worker import ToolError

async def handler(args, ctx):
    raise ToolError("UPSTREAM_TIMEOUT", "Weather API timed out")

WorkStream

Low-level helper from phrony.worker for custom registration or reconnect logic.

WorkStream

send_register, run, send_result, send_nack, in_flight_calls, close, build_handler_advertisements.

Python

import asyncio

from phrony import RuntimeClient
from phrony.gen.phrony.runtime.v1 import runtime_pb2
from phrony.worker import WorkStream, build_handler_advertisements
from phrony.worker.types import RegisteredTool


async def main() -> None:
    client = await RuntimeClient.connect()
    stream = WorkStream(client.work())

    await stream.send_register(
        runtime_pb2.WorkRegister(
            worker_id="weather-worker-1",
            handlers=build_handler_advertisements(
                [
                    RegisteredTool(
                        tool="weather.get-forecast",
                        version="1.0.0",
                        handler=lambda args, ctx: {"ok": True},
                        max_concurrency=4,
                    )
                ]
            ),
            in_flight=[
                runtime_pb2.WorkInFlightCall(call_id=call_id)
                for call_id in stream.in_flight_calls()
            ],
        )
    )

    async def on_invoke(invoke: runtime_pb2.WorkInvoke) -> None:
        stream.mark_call_executing(invoke.call_id)
        await stream.send_result(
            runtime_pb2.WorkToolResult(
                call_id=invoke.call_id,
                payload=b'{"ok": true}',
            )
        )

    stream.run(
        on_registered=lambda lease_ttl_ms: print("lease", lease_ttl_ms),
        on_invoke=lambda invoke: asyncio.create_task(on_invoke(invoke)),
    )

    await stream.close()
    await client.close()


asyncio.run(main())

Runtime client

RuntimeClient exposes every runtime unary RPC. Connect via Connect to the runtime. For agent runs, prefer Phrony.

Version and sessions

await get_version(request?)

Python

version = await client.get_version()
print(version.version, version.schema_version)

await run_session(...)

Unary session start. Pass agent_ref or bundle_ref as a string, dict, or keyword argument (mutually exclusive). input and resolved_secrets accept plain dicts.

Python

import os

response = await client.run_session(
    agent_ref="default/my-agent",
    input={"claimId": "CLM-48219"},
    resolved_secrets={"openai": os.environ["OPENAI_API_KEY"]},
)

print(response.session_id, response.status)

await run_session(bundle_ref=...)

Start a session on a deployed bundle root member.

Python

import os

response = await client.run_session(
    bundle_ref="demo/payment-desk",
    input={"message": "Pay 500 USD to Acme"},
    resolved_secrets={"stripe": os.environ["STRIPE_API_KEY"]},
)

print(response.session_id, response.status)

await list_sessions(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

response = await client.list_sessions(
    runtime_pb2.ListSessionsRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
        status="running",
        kind="",  # optional: "agent" or "bundle"
    )
)

await inspect_session(session_id=...)

Full persisted session dump: history, timeline, invocations, approvals, and delegated children.

Python

response = await client.inspect_session(session_id="sess_abc123")
# response.session.timeline — unified chronological audit view
# response.session.children — delegated child sessions (same shape, recursive)

await cancel_session(session_id=...)

Python

await client.cancel_session(session_id="sess_abc123")

await complete_session(session_id=...)

Python

await client.complete_session(session_id="sess_abc123")

Publish and deploy — agents

await publish(request)

manifest is the raw agent.yaml bytes.

Python

from pathlib import Path

from phrony.gen.phrony.runtime.v1 import runtime_pb2

manifest = Path("agent.yaml").read_bytes()
published = await client.publish(
    runtime_pb2.PublishRequest(
        manifest=manifest,
        actor="ci@example.com",
    )
)

print(published.version_id, published.content_hash)

await deploy(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.deploy(
    runtime_pb2.DeployRequest(
        agent_ref=parse_agent_ref("default/my-agent@0.2.0"),
        actor="ci@example.com",
    )
)

await rollback(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.rollback(
    runtime_pb2.RollbackRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
        to_version="0.1.0",
        actor="ops@example.com",
    )
)

await get_active_version(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

active = await client.get_active_version(
    runtime_pb2.GetActiveVersionRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
    )
)
print(active.version)

await list_deployments(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

history = await client.list_deployments(
    runtime_pb2.ListDeploymentsRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
    )
)
for deployment in history.deployments:
    print(deployment.version, deployment.deployed_at_unix_ms)

await get_agent_version(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

record = await client.get_agent_version(
    runtime_pb2.GetAgentVersionRequest(
        agent_ref=parse_agent_ref("default/my-agent@0.2.0"),
    )
)

await retire_agent_version(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.retire_agent_version(
    runtime_pb2.RetireAgentVersionRequest(
        agent_ref=parse_agent_ref("default/my-agent@0.1.0"),
    )
)

await deprecate_agent_version(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.deprecate_agent_version(
    runtime_pb2.DeprecateAgentVersionRequest(
        agent_ref=parse_agent_ref("default/my-agent@0.1.0"),
    )
)

await archive_agent(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.archive_agent(
    runtime_pb2.ArchiveAgentRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
    )
)

Publish and deploy — bundles

Bundle lifecycle RPCs mirror the CLI bundles commands. Publish requires the committed bundle.lock.json bytes alongside the bundle manifest.

await publish_bundle(request)

bundle_manifest and committed_lock are raw bytes; members lists vendored closure packages.

Python

from pathlib import Path

from phrony.gen.phrony.runtime.v1 import runtime_pb2

bundle_manifest = Path("support/bundle.yaml").read_bytes()
committed_lock = Path("support/bundle.lock.json").read_bytes()

published = await client.publish_bundle(
    runtime_pb2.PublishBundleRequest(
        bundle_manifest=bundle_manifest,
        committed_lock=committed_lock,
        members=[
            # BundleMemberPackage entries for each vendored closure member
        ],
        actor="ci@example.com",
    )
)

print(published.bundle_version_id, published.lock_hash)

await deploy_bundle(request)

bundle_ref must include @version — use parse_bundle_ref_version_required for deploy-by-hash.

Python

from phrony import parse_bundle_ref_version_required
from phrony.gen.phrony.runtime.v1 import runtime_pb2

await client.deploy_bundle(
    runtime_pb2.DeployBundleRequest(
        bundle_ref=parse_bundle_ref_version_required("demo/payment-desk@1.0.0"),
        actor="ci@example.com",
    )
)

await get_active_bundle(request)

Python

from phrony import parse_bundle_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

active = await client.get_active_bundle(
    runtime_pb2.GetActiveBundleRequest(
        bundle_ref=parse_bundle_ref("demo/payment-desk"),
    )
)
print(active.version, active.lock_hash)

await list_bundle_deployments(request)

Python

from phrony import parse_bundle_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

history = await client.list_bundle_deployments(
    runtime_pb2.ListBundleDeploymentsRequest(
        bundle_ref=parse_bundle_ref("demo/payment-desk"),
    )
)
for deployment in history.deployments:
    print(deployment.version, deployment.lock_hash, deployment.deployed_at)

Catalog

await list_agents(request?)

Python

response = await client.list_agents()
for agent in response.agents:
    print(agent.namespace, agent.name)

await list_agent_versions(request)

Python

from phrony import parse_agent_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

response = await client.list_agent_versions(
    runtime_pb2.ListAgentVersionsRequest(
        agent_ref=parse_agent_ref("default/my-agent"),
    )
)

await list_bundles(request?)

Python

response = await client.list_bundles()
for bundle in response.bundles:
    print(bundle.namespace, bundle.name)

await list_bundle_versions(request)

Python

from phrony import parse_bundle_ref
from phrony.gen.phrony.runtime.v1 import runtime_pb2

response = await client.list_bundle_versions(
    runtime_pb2.ListBundleVersionsRequest(
        bundle_ref=parse_bundle_ref("demo/payment-desk"),
    )
)

Approvals

await get_approval(approval_id=...)

Python

approval = await client.get_approval(approval_id="appr_xyz")

await list_approvals(...)

Filter by status, session_id, route, agent_namespace, and agent_name. Omit filters to list all.

Python

response = await client.list_approvals(
    status="pending",
    session_id="",
    route="",
)

await decide_approval(...)

Python

from phrony import ApprovalDecision

await client.decide_approval(
    approval_id="appr_xyz",
    decision=ApprovalDecision.APPROVAL_DECISION_APPROVE,
    comment="Looks good",
    args={"city": "Oslo"},
    comprehension_acknowledged=True,
    actor="ops@example.com",
)

Streams

work(metadata?)

Raw grpc.aio duplex stream — prefer Worker or WorkStream.

Python

raw = client.work()  # grpc.aio stream — prefer Worker

run_session_interactive(metadata?)

Returns InteractiveSession — see Interactive sessions.

Python

session = client.run_session_interactive()

Typical usage wraps the client in try/finally:

Pattern

Python

import asyncio

from phrony import RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    try:
        # examples below
        pass
    finally:
        await client.close()


asyncio.run(main())

Utilities

Agent references

parse_agent_ref(string)

Python

from phrony import format_agent_ref, parse_agent_ref

ref = parse_agent_ref("default/my-agent@0.2.0")
# {"namespace": "default", "name": "my-agent", "version": "0.2.0"}

format_agent_ref(ref)  # "default/my-agent@0.2.0"

format_agent_ref(ref)

Python

from phrony import format_agent_ref

format_agent_ref({
    "namespace": "default",
    "name": "my-agent",
    "version": "0.2.0",
})  # "default/my-agent@0.2.0"

Bundle references

parse_bundle_ref(string)

Version may be semver or a lock hash (sha256:…).

Python

from phrony import format_bundle_ref, parse_bundle_ref

semver = parse_bundle_ref("demo/payment-desk@1.0.0")
# {"namespace": "demo", "name": "payment-desk", "version": "1.0.0"}

lock_hash = parse_bundle_ref("demo/payment-desk@sha256:abc…")
format_bundle_ref(lock_hash)  # "demo/payment-desk@sha256:abc…"

format_bundle_ref(ref)

Python

from phrony import format_bundle_ref

format_bundle_ref({
    "namespace": "demo",
    "name": "payment-desk",
    "version": "1.0.0",
})  # "demo/payment-desk@1.0.0"

parse_bundle_ref_version_required(string)

Same as parse_bundle_ref but rejects empty @version — use for deploy_bundle.

Python

from phrony import parse_bundle_ref_version_required

# Deploy flows require an explicit @version (semver or lock hash)
ref = parse_bundle_ref_version_required("demo/payment-desk@sha256:abc…")

JSON bytes

json_bytes / parse_json_bytes / json_bytes_map / resolved_secrets_map

Encode and decode proto bytes fields that carry UTF-8 JSON. resolved_secrets_map stores raw UTF-8 secret values (not JSON-encoded).

Python

import os

from phrony import json_bytes, json_bytes_map, parse_json_bytes, resolved_secrets_map

input_bytes = json_bytes({"claimId": "CLM-48219"})
parsed = parse_json_bytes(input_bytes)
secrets = resolved_secrets_map({"openai": os.environ["OPENAI_API_KEY"]})
encoded_map = json_bytes_map({"openai": os.environ["OPENAI_API_KEY"]})

Errors

PhronyRuntimeError

Python

from phrony import PhronyRuntimeError

try:
    await client.get_version()
except PhronyRuntimeError as err:
    print(err.grpc_code, err.action, err.details)

wrap_rpc_error(action, err)

Python

from phrony import wrap_rpc_error

try:
    await some_grpc_call()
except Exception as err:
    raise wrap_rpc_error("my action", err) from err

AgentSessionError

Python

import asyncio

from phrony import AgentSessionError, Phrony


async def main() -> None:
    try:
        phrony = await Phrony.connect()
        await phrony.agent("default/my-agent").run(input={})
    except AgentSessionError as err:
        print(err.session_id, err)


asyncio.run(main())

AgentRefParseError

Python

from phrony import AgentRefParseError, parse_agent_ref

try:
    parse_agent_ref("my-agent")  # missing namespace/
except AgentRefParseError as err:
    print(err)

BundleRefParseError

Python

from phrony import BundleRefParseError, parse_bundle_ref

try:
    parse_bundle_ref("payment-desk")  # missing namespace/
except BundleRefParseError as err:
    print(err)

ToolError

Python

from phrony.worker import ToolError

async def handler(args, ctx):
    raise ToolError("UPSTREAM_TIMEOUT", "Weather API timed out")

Constants and helpers

SDK_VERSION, DEFAULT_MAX_CONCURRENCY, handler_key, heartbeat_interval_ms

Python

from phrony import (
    DEFAULT_MAX_CONCURRENCY,
    SDK_VERSION,
    handler_key,
    heartbeat_interval_ms,
)

print(SDK_VERSION)
print(handler_key("weather.get-forecast", "1.0.0"))  # weather.get-forecast@1.0.0
print(heartbeat_interval_ms(30_000))  # 15000

Generated types

Import typed shapes from phrony for refs, interactive events, and common responses. Unary methods with keyword-arg convenience include run_session, cancel_session, complete_session, inspect_session, get_approval, decide_approval, and list_approvals. For publish, deploy, and catalog RPCs, pass protobuf request messages from phrony.gen.phrony.runtime.v1 — nested agent_ref and bundle_ref fields accept dicts from parse_agent_ref / parse_bundle_ref.

Type imports

Python

from phrony import (
    AgentRef,
    ApprovalDecision,
    BundleRef,
    GetVersionResponse,
    InteractiveEvent,
    InteractiveSessionAttachOptions,
    InteractiveSessionStartOptions,
    InteractiveToolApprovalOptions,
    RunSessionResponse,
)

agent_ref: AgentRef = {
    "namespace": "default",
    "name": "my-agent",
    "version": "",
}

# Pass a string ref, dict, or keyword args to run_session / InteractiveSession.start.
# Convenience kwargs: run_session, cancel_session, complete_session, inspect_session,
# get_approval, decide_approval, list_approvals.
# For other unary RPCs, pass protobuf requests from phrony.gen.phrony.runtime.v1.

Commonly used types

AgentRef, BundleRef, InteractiveEvent, InteractiveSessionStartOptions, ApprovalDecision, GetVersionResponse, RunSessionResponse