Rungate Python SDK

The official Python SDK for the Rungate governance proxy.

Installation

pip install rungate

Quick Start

from rungate import Rungate

client = Rungate(
    api_key="rg_sk_...",               # or set RUNGATE_API_KEY env var
    base_url="http://localhost:8080",   # or set RUNGATE_BASE_URL env var
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

print(response.choices[0]["message"]["content"])
print(f"Cost: ${response.rungate.cost_usd}")
print(f"Budget remaining: ${response.rungate.budget_remaining_usd}")

Run Lifecycle

Track agent executions with explicit run management:

with client.run(name="research-task", metadata={"customer": "abc"}) as run:
    # All LLM calls automatically tracked under this run
    response = run.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Summarize this document"}],
    )

    # Check budget in real-time
    print(f"Budget remaining: ${run.budget.remaining_usd}")
    print(f"Total cost so far: ${run.budget.total_cost}")

    # Log non-LLM tool calls
    run.log_tool_call(
        name="web_search",
        request={"query": "rungate governance proxy"},
        response={"results_count": 5},
        latency_ms=145,
    )

    # Make more LLM calls — budget accumulates
    response2 = run.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Now analyze the results"}],
    )

# Run auto-completes on exit, auto-fails on exception

Streaming

Low-level (raw chunks)

with client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
) as stream:
    for chunk in stream:
        delta = chunk["choices"][0]["delta"]
        if "content" in delta:
            print(delta["content"], end="")

    print(f"\nCost: ${stream.rungate.cost_usd}")

High-level (text stream)

with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="")

    completion = stream.get_final_completion()
    print(f"\nCost: ${stream.rungate.cost_usd}")

Governance-Aware Error Handling

from rungate import (
    Rungate,
    GovernanceError,
    BudgetExceededError,
    ToolBlockedError,
    ApprovalRequiredError,
    RateLimitedError,
)

try:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}],
    )
except BudgetExceededError as e:
    print(f"Over budget: {e.details}")
    print(f"Run: {e.run_id}")
except ToolBlockedError as e:
    print(f"Tool blocked: {e.details['tool']}")
except ApprovalRequiredError as e:
    print(f"Tools need approval:")
    for req in e.approval_requests:
        print(f"  - {req['tool']} (ID: {req['id']})")
except RateLimitedError as e:
    print(f"Rate limited: {e.details}")
except GovernanceError as e:
    # Catch-all for any governance block
    print(f"Governance: {e.governance_code}")

Admin Client

Manage Rungate resources programmatically:

from rungate import Admin

admin = Admin(
    token="rg_adm_...",                # or set RUNGATE_ADMIN_TOKEN env var
    base_url="http://localhost:8080",
)

# Manage agents
agents = admin.agents.list()
new_agent = admin.agents.create(name="my-agent", policy_id="...")

# Manage policies
policies = admin.policies.list()
admin.policies.update(policy_id, budget={"hard_limit_usd": 50.0})

# Approve tool calls (supervisor agent pattern)
pending = admin.approvals.list(status="pending")
for request in pending["data"]:
    admin.approvals.approve(request["id"])

# Manage alert rules
admin.alert_rules.create(
    name="high-cost-alert",
    condition_type="cost_threshold",
    condition_config={"threshold_usd": 100, "window_seconds": 3600},
)

# Analytics — cost breakdown
costs = admin.analytics.costs(group_by="agent", since="2026-03-01T00:00:00Z")

# Analytics — latency percentiles (p50/p95/p99)
latency = admin.analytics.latency(group_by="provider", granularity="hour")

# Analytics — error rates and status breakdown
errors = admin.analytics.errors(group_by="provider")

# Analytics — request and token throughput
throughput = admin.analytics.throughput(group_by="model")

# Analytics — provider reliability (success rate, avg latency)
reliability = admin.analytics.provider_reliability()

# Analytics — run completion stats
runs = admin.analytics.runs(granularity="day")

# Analytics — top-N rankings
top_spenders = admin.analytics.top(metric="spend", entity="agent", limit=10)

# Analytics — overview dashboard stats
overview = admin.analytics.overview()

# Provider health (circuit breaker status)
health = admin.providers.health()

Async Support

from rungate import AsyncRungate, AsyncAdmin

async with AsyncRungate(api_key="rg_sk_...") as client:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )

async with AsyncAdmin(token="rg_adm_...") as admin:
    agents = await admin.agents.list()

Links

• Rungate Server
• API Documentation