phrony 0.0.0

Python SDK for the Phrony runtime (gRPC).

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