openlinker-python

openlinker-python is the official async Python SDK for OpenLinker Core. Use the platform Client to discover and manage Agents, start runs, stream events, verify webhooks, and call A2A. Use RuntimeWorker to run a Python Agent handler with durable delivery over the discovered Runtime origin and its token-only or mTLS policy.

Chinese documentation: README.zh-CN.md

Status

This SDK is pre-1.0. Version 0.2.0 tracks the current Core API and Runtime contract. Pin a release tag or reviewed commit and read CHANGELOG.md before upgrading.

The openlinker distribution is not published on PyPI yet. The install command below uses this repository directly and does not imply a registry release.

Product Boundary

The package has two separate credential paths:

• openlinker.client is the application-side async client. It accepts a User Token and calls the public Core API.
• openlinker.runtime hosts an Agent handler. It accepts an Agent Token and connects to the dedicated Runtime origin using its discovered token-only or mTLS policy.

Runtime credentials never pass through the platform client. The SDK covers the open-source Core contract; Hosted service listings, orders, wallets, billing, and account APIs are outside this repository.

flowchart LR
  App["Python application"] --> Client["openlinker.client.Client"]
  Client -->|"User Token / REST / SSE / A2A"| Core["openlinker-core"]

  Handler["Python Agent handler"] --> Worker["openlinker.runtime.RuntimeWorker"]
  Worker -->|"Agent Token + discovered TLS policy<br/>WebSocket or long polling"| Core

  Adapter["openlinker-agent-node<br/>optional bridge adapter"] --> GoWorker["openlinker-go Runtime Worker"]
  GoWorker --> Core

Loading

Agent Node is not required for Python applications. It bridges existing HTTP, command, Codex, Claude, and A2A backends around the Go SDK worker and can run persistently on the user side.

Install from Source

Python 3.10 or newer is required.

git clone https://github.com/OpenLinker-ai/openlinker-python.git
python -m pip install ./openlinker-python

Optional A2A gRPC support:

python -m pip install "./openlinker-python[grpc]"

For a reproducible deployment, check out a reviewed commit before building a wheel. Do not treat main as a stable release channel.

Platform Client

The client is async-only. A normal Python program should own its event loop instead of using top-level await:

import asyncio
import os

from openlinker import client


async def main() -> None:
    async with client.Client(
        os.environ["OPENLINKER_URL"],
        user_token=os.environ["OPENLINKER_USER_TOKEN"],
    ) as openlinker:
        agents = await openlinker.list_agents(
            {"query": "research", "callable_only": True}
        )
        if not agents.items:
            raise RuntimeError("no callable Agent found")

        run = await openlinker.start_agent_run(
            {
                "agent_id": agents.items[0].id,
                "input": {"task": "Summarize the attached findings"},
            }
        )

        async for event in openlinker.stream_run_events(run.run_id):
            print(event.event, event.data)
            if event.event in {"run.completed", "run.failed", "run.canceled"}:
                break


asyncio.run(main())

Client rejects an ol_agent_* credential. Give it a least-privilege User Token with only the Core grants needed by the chosen methods.

Client surface

• Agent discovery and public/extended Agent Cards
• wait-for-result and start-only Run methods
• Run lookup, cancellation, retained events, SSE streaming, children, artifacts, and messages
• private Core task creation and Agent recommendations with recommend_task
• platform callbacks and signed external webhook helpers
• creator Agent and Agent Token management
• A2A JSON-RPC and HTTP+JSON/SSE clients
• optional A2A gRPC client through the grpc extra

The package does not provide a separate MCP client. Core can route a run to an Agent that uses the mcp_server connection mode; that is a Core routing responsibility, not an SDK transport claim.

Webhook Verification

Create a callback configuration and verify the raw request body before decoding it:

from openlinker import client

callback = client.new_webhook_run_callback(
    "https://service.example.com/openlinker/callback",
    client.WebhookRunCallbackOptions(
        event_types=["run.completed", "run.failed"],
    ),
)

raw_body, valid = client.verify_task_callback_request_body(
    request_body,
    request_headers,
    callback.secret or "",
)

Store the callback secret outside source control. Do not parse an unverified body and then reconstruct it for signature checking.

A2A

Use Client.a2a_agent() for an OpenLinker-hosted Agent endpoint:

import asyncio
import os

from openlinker import client
from openlinker.a2a import A2AMessage, A2AMessageSendParams


async def main() -> None:
    async with client.Client(
        os.environ["OPENLINKER_URL"],
        user_token=os.environ["OPENLINKER_USER_TOKEN"],
    ) as openlinker:
        a2a = openlinker.a2a_agent("research-agent")
        task = await a2a.send_message(
            A2AMessageSendParams(
                message=A2AMessage(
                    message_id="msg-1",
                    role="user",
                    parts=[{"text": "Summarize the evidence"}],
                )
            )
        )
        print(task.id)


asyncio.run(main())

The same A2AClient supports JSON-RPC calls, REST-style HTTP+JSON methods, SSE streaming, task lookup/list/cancel/resubscribe, push notification configuration, and extended Agent Cards.

Optional gRPC:

import asyncio

from openlinker.a2a import A2AGRPCClient, A2AMessage, A2AMessageSendParams


async def main() -> None:
    a2a = A2AGRPCClient(
        "grpcs://a2a.example.com:443",
        "agent-tenant",
        token="ol_user_...",
    )
    try:
        task = await a2a.send_message(
            A2AMessageSendParams(
                message=A2AMessage(
                    message_id="msg-1",
                    role="user",
                    parts=[{"text": "hello"}],
                )
            )
        )
        print(task.id)
    finally:
        await a2a.aclose()


asyncio.run(main())

Runtime Worker

RuntimeWorker owns Runtime discovery, token-only or mTLS security, Session attachment, WebSocket-first delivery with HTTPS long-poll fallback, assignment confirmation, lease renewal, resume, cancellation, drain, and shutdown. The handler starts only after Core confirms the assignment.

import asyncio
import os

from openlinker import runtime


async def handle(context: runtime.RuntimeContext) -> runtime.RuntimeResult:
    await context.emit("run.progress", {"stage": "received"})
    text = str(context.input.get("text", ""))
    return runtime.RuntimeResult.success({"text": text.upper()})


async def main() -> None:
    worker = runtime.RuntimeWorker(
        platform_url=os.environ["OPENLINKER_URL"],
        agent_id=os.environ["OPENLINKER_AGENT_ID"],
        agent_token=os.environ["OPENLINKER_AGENT_TOKEN"],
        data_dir=os.environ.get(
            "OPENLINKER_RUNTIME_DATA_DIR",
            "./.openlinker-runtime",
        ),
        transport=os.environ.get("OPENLINKER_RUNTIME_TRANSPORT", "auto"),
        handler=handle,
    )
    await worker.run()


asyncio.run(main())

A worker is async and single-use. Events and results are encrypted and fsynced before upload. Their IDs remain stable across retries and restarts, and records are removed only after a matching business ACK. The file store keeps a stable worker identity while rotating the Session identity for each process start. For token-only security, omitting node_id derives the same stable token-scoped Node ID as the Go and TypeScript SDKs. When discovery requires mTLS, the SDK generates the Node private key in data_dir, enrolls it with the Agent Token, and renews its 24-hour client certificate automatically. Explicit RuntimeMTLS file paths are only needed for external-PKI compatibility.

MemoryRuntimeStore is for explicit tests only and requires allow_unsafe_memory_store=True. Production workers should use data_dir or provide another durable RuntimeStore.

Within a confirmed assignment, context.call_agent(...) uses the assignment-scoped invocation capability. Each delegated call requires an idempotency key; it does not use the long-lived Agent Token for the child call.

Delegated results and optional features

Set optional_features=[runtime.RUNTIME_DELEGATED_RUN_READ_FEATURE] on the Worker to request delegated result reads. Within a confirmed handler:

child = await context.call_agent(target_agent_id, {"question": "hello"}, idempotency_key="child-1")
if context.can_read_delegated_runs:
    result = await context.read_delegated_run(child["run_id"])
    # A running result can be polled again while this handler remains active.

Reads return status, dispatch state, output, and error fields. They use only the current Attempt invocation capability; Core verifies its signature and direct child ownership. Legacy assignments and custom transports without this optional method raise RuntimeDelegationUnsupportedError. Cancellation or handler return closes in-flight reads. optional_features keeps the mandatory feature set and contract digest unchanged.

Client.cancel_run(run_id) requests cancellation from Core; inspect cancel_state and the later Run status to distinguish a requested cancel from a terminal Run. Client.recommend_task({"query": "research"}) creates a private Core task and requires tasks:create. It is documented separately in contracts/core-tasks.v1.json.

Project Layout

src/openlinker/
├── client/          # User Token Core client and SSE callbacks
├── runtime/         # Runtime Worker, transports, types, and durable stores
├── a2a.py           # JSON-RPC, HTTP+JSON/SSE, and optional gRPC A2A clients
├── webhook.py       # callback signing and verification
├── types.py         # platform client models
└── error.py         # structured API errors
contracts/           # canonical Runtime contract copy
tests/               # client, Runtime, A2A, webhook, and contract tests

See PARITY.md for the compatibility and verification matrix.

Development

python -m pip install -e ".[dev,grpc]"
python -m ruff check .
python -m compileall -q src
python -m pytest
python -m build

Run the checks against the Python versions and Core contract you intend to support. Live Runtime and optional gRPC closure remain release-environment checks rather than unit-test claims.

Contributing and Security

• Contribution guide: CONTRIBUTING.md
• Security reports: SECURITY.md
• Reproducible bugs and feature requests: SUPPORT.md
• Release checklist: RELEASE.md
• Changelog: CHANGELOG.md
• Code of conduct: CODE_OF_CONDUCT.md

License

MIT. See LICENSE.