phrony

Python client for the Phrony runtime over gRPC. Use it to run agents, open interactive sessions, and register tool workers.

This package targets Python 3.10+ and speaks the runtime API defined in runtime/proto/phrony/runtime/v1/runtime.proto. It is not the cloud HTTP control-plane client.

Install

pip install phrony

Runtime address

By default the SDK dials 127.0.0.1:7777. Override with the environment variable or options on each entrypoint:

export PHRONY_RUNTIME_ADDR=127.0.0.1:7777

Start the runtime locally (see the runtime repo) before calling the SDK.

Quick start

Run an agent

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

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


asyncio.run(main())

Agent references use namespace/name or namespace/name@version. Set wait=False on run() to start a session via unary RunSession and return immediately with the session id.

Register a tool worker

import asyncio
import os

from phrony.worker import Worker, ToolError


worker = Worker(
    worker_id="weather-worker-1",
    runtime_addr=os.environ.get("PHRONY_RUNTIME_ADDR", "127.0.0.1:7777"),
)


@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"]}


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


asyncio.run(main())

Workers connect on the bidirectional Work stream: register handlers, heartbeats, invoke/result, and graceful shutdown. Register all tools before connect(). You can also call register_tool(...) imperatively for programmatic registration.

Low-level gRPC client

For full control over unary RPCs and streams:

import asyncio

from phrony import RuntimeClient


async def main() -> None:
    client = await RuntimeClient.connect()
    version = await client.get_version()
    print(version.version)

    session = client.run_session_interactive()
    await session.start(
        agent_ref="default/my-agent",
        input={"question": "hello"},
    )

    async for event in session.events():
        if event["type"] == "session_started":
            print(event["session_id"], event["agent_version_id"])
        if event["type"] == "text_delta":
            print(event["delta"], end="", flush=True)
        if event["type"] == "approval_required":
            await session.decide_tool_approval(
                approval_id=event["approval_id"],
                approved=True,
            )
        if event["type"] == "completed":
            print(event.get("output"))

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


asyncio.run(main())

Proto bytes fields that carry JSON (input, args, payload, and similar) are handled via json_bytes and parse_json_bytes from the main export. For advanced use you can still import generated protobuf types from phrony.gen.phrony.runtime.v1.

Development

This repo is managed with uv. Prerequisites:

• uv
• protobuf (protoc) when regenerating stubs — e.g. brew install protobuf

uv sync
uv run python scripts/generate_proto.py   # regenerate src/phrony/gen from ../runtime/proto
uv run pytest
uv run ruff check src tests
uv run mypy
uv build

Integration tests (require a running runtime on the configured address):

PHRONY_INTEGRATION=1 uv run pytest tests/integration

From the runtime repo you can bring up a local daemon and point PHRONY_RUNTIME_ADDR at it.

Scope and limitations

• Local dev uses insecure gRPC by default; pass channel credentials in client options when you add TLS later.
• Agent manifests and compile/publish flows live outside this package.

License

MIT