Runtime Client

Switch between Python and TypeScript code examples using the language tabs.

The ZabyRuntime client is for interacting with deployed agents at runtime. Unlike Zaby, it uses a JWT token for authentication rather than an API key. See Runtime Tokens for generating a JWT from zaby.runtime_tokens.create().

Python
TypeScript

from zaby import ZabyRuntime

runtime = ZabyRuntime(token="eyJhbGciOi...")

import { ZabyRuntime } from "@zaby-ai/sdk";

const runtime = new ZabyRuntime({ token: "eyJhbGciOi..." });

For non-production environments, use configureZaby() to set the API origin before creating ZabyRuntime:

Python
TypeScript

from zaby import ZabyGlobalConfig, configure_zaby
configure_zaby(ZabyGlobalConfig(api_origin="http://localhost:9080"))

import { configureZaby } from "@zaby-ai/sdk";
configureZaby({ apiOrigin: "http://localhost:9080" });

Create a Runtime Token

Generate a JWT from zaby.runtime_tokens.create() or zaby.runtimeTokens.create()

Agents Client

Create, publish, and deploy agents.

Start a run

input

string

required

The user’s message to the agent

requestId

string

Idempotency key for safe retries — use a unique value per request

Python
TypeScript

run = await runtime.runs.start({
    "input": "What can you help me with?",
    "requestId": "my-req-001",
})

const run = await runtime.runs.start({
  input: "What can you help me with?",
  requestId: "my-req-001",
});

runId

string

Unique run identifier

status

string

Run status: "COMPLETED", "FAILED", or "PENDING"

finalOutput

string

Agent’s final response text (present when status is COMPLETED)

{
  "workflowId": "04a3bca0-...",
  "runId": "0bbce854-...",
  "deploymentId": "9f2b235d-...",
  "status": "COMPLETED",
  "latestSeq": 15,
  "finalOutput": "Hello! How can I assist you today?",
  "agentSessionId": "ff44fb37-...",
  "externalAppId": "2abd6574-..."
}

Stream agent responses

Stream real-time agent output via Server-Sent Events.

Python
TypeScript

async for event in runtime.runs.stream(run_id):
    if event.event == "text":
        print(event.data["content"], end="", flush=True)
    elif event.event == "tool_call":
        print(f"\n[Calling tool: {event.data['toolName']}]")
    elif event.event == "RunFinished":
        print(f"\n\nRun complete: {event.data}")

for await (const event of runtime.runs.stream(run_id)) {
  if (event.event === "text") {
    process.stdout.write(event.data.content);
  } else if (event.event === "tool_call") {
    console.log(`\n[Calling tool: ${event.data.toolName}]`);
  } else if (event.event === "RunFinished") {
    console.log(`\n\nRun complete: ${JSON.stringify(event.data)}`);
  }
}

Event types

Event	Fires when
`text`	Agent produces text output
`tool_call`	Agent invokes an MCP tool
`tool_result`	Tool returns a result
`thinking`	Agent is processing
`RunFinished`	Run completes

SseEvent fields

event

string

Event type identifier

data

any

Parsed JSON payload

string

Optional event ID for tracking

Retrieve events

Fetch stored events for a completed run:

limit

number

Maximum events to return (default: 50)

Python
TypeScript

events = await runtime.runs.events(run_id, {"limit": 100})

const events = await runtime.runs.events(run_id, { limit: 100 });

Approvals

Respond to approval requests during a run:

Python
TypeScript

await runtime.approvals.approve(run_id, approval_id)
await runtime.approvals.reject(run_id, approval_id)

await runtime.approvals.approve(run_id, approval_id);
await runtime.approvals.reject(run_id, approval_id);

Feedback

rating

number

required

Rating score from 1 to 5

comment

string

Optional free-text feedback

Python
TypeScript

await runtime.feedback.create(run_id, {
    "rating": 5,
    "comment": "Excellent response!",
})

await runtime.feedback.create(run_id, {
  rating: 5,
  comment: "Excellent response!",
});

Complete example

Requires the full agent lifecycle: create agent → publish → deploy → bind external app → create quota policy → generate runtime token.

Python
TypeScript

import asyncio
from zaby import ZabyGlobalConfig, configure_zaby, ZabyRuntime

# For local/self-hosted:
# configure_zaby(ZabyGlobalConfig(api_origin="http://localhost:9080"))

async def chat_with_agent():
    runtime = ZabyRuntime(token="eyJhbGciOi...")

    run = await runtime.runs.start({
        "input": "What's the weather in London?",
    })
    run_id = run["runId"]

    async for event in runtime.runs.stream(run_id):
        if event.event == "text":
            print(event.data["content"], end="", flush=True)
        elif event.event == "RunFinished":
            print()

    await runtime.feedback.create(run_id, {"rating": 5})

asyncio.run(chat_with_agent())

import { configureZaby, ZabyRuntime } from "@zaby-ai/sdk";

// For local/self-hosted: configureZaby({ apiOrigin: "http://localhost:9080" })

async function chatWithAgent() {
  const runtime = new ZabyRuntime({ token: "eyJhbGciOi..." });

  const run = await runtime.runs.start({
    input: "What's the weather in London?",
  });
  const run_id = run.runId;

  for await (const event of runtime.runs.stream(run_id)) {
    if (event.event === "text") {
      process.stdout.write(event.data.content);
    } else if (event.event === "RunFinished") {
      console.log();
    }
  }

  await runtime.feedback.create(run_id, { rating: 5 });
}

chatWithAgent();

Agents & Provisioning

Full agent lifecycle reference.

Testing

Mock SSE streaming with MockTransport.

​Runtime Client

Create a Runtime Token

Agents Client

​Start a run

​Stream agent responses

​Event types

​SseEvent fields

​Retrieve events

​Approvals

​Feedback

​Complete example

Agents & Provisioning

Testing

Runtime Client

Start a run

Stream agent responses

Event types

SseEvent fields

Retrieve events

Approvals

Feedback

Complete example