syrin-sdk 1.2.0

AI agent observability and control SDK for Syrin

Syrin SDK for Python

Observability, remote config, and governance for AI agents — one import, one init() call, zero changes to your existing code.

📖 New here? Start with the Complete Onboarding Guide — covers every feature from installation to production in one place.

---

What Syrin gives you

Capability	What it means
Session timeline	Every LLM call, cost, latency, and custom event grouped by user and run
Remote config	Change model, temperature, prompts live from the dashboard — no redeploy
Governance	Stop or constrain agents at runtime from the backend
Checkpoints	Save and restore conversation state for recovery flows
Custom events	Emit structured log entries that appear on the session timeline
OpenTelemetry	Standard gen_ai.* spans + syrin.* extensions, works with any OTLP backend

---

Install

pip install syrin-sdk

Requires Python 3.11+. The only required dependency is httpx; OpenTelemetry is optional.

---

Setup — 2 lines

import syrin_sdk

syrin_sdk.init(api_key="syrin_...")

That is the entire setup. Every OpenAI call in your process is now instrumented automatically.

---

Core concepts

Sessions — group events by user and time window

Wrap each user request in a context() call. Everything inside — LLM calls, costs, custom logs — appears together on the dashboard timeline.

with syrin_sdk.context(user_id="alice", window="day") as ctx:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
    )
    print(ctx.session_id)  # u:alice:2026-04-19

Session IDs are deterministic within the window, so all of Alice's requests today share the same session — giving you one continuous conversation history per user per day.

Pattern	Session ID
context()	ses_a1b2c3 (random UUID)
context(user_id="alice", window="hour")	u:alice:2026-04-19T14
context(user_id="alice", window="day")	u:alice:2026-04-19
context(user_id="alice", window="week")	u:alice:2026-W16
context(user_id="alice", window="month")	u:alice:2026-04
context(user_id="alice", window="forever")	u:alice
context(key="batch-etl", window="day")	k:batch-etl:2026-04-19

# Async — same API
async with syrin_sdk.context(user_id="alice", window="day") as ctx:
    response = await client.chat.completions.create(...)

---

Agent scoping — tag events with the agent that produced them

Pass agent= to context() to tag every event inside with that agent:

with syrin_sdk.context(user_id="alice", agent="researcher", window="day") as ctx:
    response = client.chat.completions.create(...)

For multi-agent workflows, use workflow() to group agent runs together:

with syrin_sdk.workflow("research-pipeline") as ctx:
    with syrin_sdk.context(agent="planner") as c:
        plan = call_llm(plan_messages)
    with syrin_sdk.context(agent="executor") as c:
        result = call_llm(exec_messages)

---

Remote config — cfg()

Declare any parameter as remotely configurable. Push overrides from the dashboard and they take effect on the next call — no redeploy, no restart.

response = client.chat.completions.create(
    model=syrin_sdk.cfg("llm.model", "gpt-4o"),
    temperature=syrin_sdk.cfg("llm.temperature", 0.7, ge=0.0, le=2.0),
    max_tokens=syrin_sdk.cfg("llm.max_tokens", 1024),
    messages=[
        {"role": "system", "content": syrin_sdk.cfg("prompt.system", "You are helpful.", multiline=True)},
        {"role": "user", "content": user_message},
    ],
)

• Key format: "section.field" — sections appear as accordion groups in the dashboard
• Default: used until you push an override
• Constraints: ge, le, enum validated before delivery
• Priority: governance override → local configure() → remote push → default

---

Custom events — log()

Emit structured events that appear on the session timeline with timestamp, level, and metadata.

syrin_sdk.log("Retrieved 42 documents", metadata={"collection": "kb", "latency_ms": 45})
syrin_sdk.log("Cost budget at 80%", level="warning")
syrin_sdk.log("Tool call failed", level="error", metadata={"tool": "web_search", "error": str(e)})

Levels: "debug", "info" (default), "warning", "error".

---

Governance — handle backend stops

The backend can stop an agent mid-run when a governance rule fires (cost exceeded, loop detected, etc.). Enable it and catch GovernanceStopError:

syrin_sdk.init(
    api_key="syrin_...",
    governance={"allow_stop": True},   # opt in to destructive actions
)

from syrin_sdk import GovernanceStopError

try:
    response = client.chat.completions.create(...)
except GovernanceStopError as e:
    logger.warning("Agent stopped: %s (incident: %s)", e.reason, e.incident_id)
    return {"error": "request_blocked", "reason": e.reason}

---

Multi-agent apps — AgentHandle

For apps with multiple agents, use AgentHandle to declare each agent's config fields once and scope calls precisely.

import syrin_sdk

sdk = syrin_sdk.init(api_key="syrin_...")

# ── Declare agents and their configurable fields ──────────────────────────────
researcher = sdk.agent("researcher")
researcher.field("llm.temperature", 0.3, ge=0.0, le=2.0, label="Temperature")
researcher.field("llm.model", "gpt-4o", label="Model")
researcher.field("prompt.system_prompt", "Research thoroughly.", multiline=True)

writer = sdk.agent("writer")
writer.field("llm.temperature", 0.7, ge=0.0, le=2.0)
writer.field("output.format", "markdown", enum=["markdown", "plain", "html"])

# ── Session + agent scope in one call ─────────────────────────────────────────
with researcher.session(user_id="alice", window="day") as ctx:
    temp   = researcher.cfg("llm.temperature", 0.3)
    model  = researcher.cfg("llm.model", "gpt-4o")
    prompt = researcher.cfg("prompt.system_prompt", "Research thoroughly.")

    response = client.chat.completions.create(
        model=model,
        temperature=temp,
        messages=[
            {"role": "system", "content": prompt},
            {"role": "user", "content": query},
        ],
    )

# ── Agent scope only (no session) ─────────────────────────────────────────────
with writer.run() as ctx:
    fmt = writer.cfg("output.format", "markdown")
    # ... call LLM ...

# ── Multi-agent workflow ───────────────────────────────────────────────────────
with sdk.workflow("research-and-write") as wf:
    with researcher.run() as r:
        research_result = call_llm(research_prompt)
    with writer.run() as w:
        final_output = call_llm(write_prompt)

Each agent's fields appear as a separate group in the Syrin dashboard, independently controllable.

---

Decorator — @traced()

For function-based agents, use @traced() instead of with context():

@syrin_sdk.traced(agent="researcher", window="day")
def research(query: str, user_id: str) -> str:
    return chain.invoke({"query": query})

# Async works the same way
@syrin_sdk.traced(agent="summarizer")
async def summarize(text: str) -> str:
    return await async_chain.ainvoke({"text": text})

---

Full example — Flask chat server

import syrin_sdk
from syrin_sdk import GovernanceStopError
from openai import OpenAI
from flask import Flask, request, jsonify

# ── Init ──────────────────────────────────────────────────────────────────────
sdk = syrin_sdk.init(
    api_key="syrin_...",
    agent_id="chat-agent",
    governance={"allow_stop": True},
)

chat = sdk.agent("chat")
chat.field("llm.model", "gpt-4o", label="Model")
chat.field("llm.temperature", 0.7, ge=0.0, le=2.0, label="Temperature")
chat.field("llm.max_tokens", 1024, ge=1, le=8192)
chat.field("prompt.system", "You are a helpful assistant.", multiline=True)

client = OpenAI()
app = Flask(__name__)

# ── Routes ────────────────────────────────────────────────────────────────────
@app.route("/chat", methods=["POST"])
def chat_endpoint():
    body = request.json
    user_id = body.get("user_id", "anonymous")
    messages = body.get("messages", [])

    try:
        with chat.session(user_id=user_id, window="day") as ctx:
            system_prompt = chat.cfg("prompt.system", "You are a helpful assistant.")
            full_messages = [{"role": "system", "content": system_prompt}] + messages

            response = client.chat.completions.create(
                model=chat.cfg("llm.model", "gpt-4o"),
                temperature=chat.cfg("llm.temperature", 0.7),
                max_tokens=chat.cfg("llm.max_tokens", 1024),
                messages=full_messages,
            )
            reply = response.choices[0].message.content
            syrin_sdk.log("Chat completed", metadata={"turns": len(messages)})

    except GovernanceStopError as e:
        return jsonify({"error": "blocked", "reason": e.reason}), 503

    return jsonify({"reply": reply, "session_id": ctx.session_id})


@app.route("/health")
def health():
    ok = sdk.health_check()
    return jsonify({"ok": ok}), 200 if ok else 503

---

All init() options

sdk = syrin_sdk.init(
    api_key="syrin_...",          # Required — from dashboard Settings
    agent_id="my-agent",          # Default agent ID for un-scoped calls
    backend_url="https://...",    # Default: Syrin cloud (https://api.syrin.dev)
    offline=False,                # True = no network calls (local dev / CI)
    capture_content=False,        # True = record prompt/response text (check PII policy)
    otel_exporter="none",         # "none" | "console" | "otlp"
    otel_endpoint="http://...",   # OTLP endpoint (Jaeger, Tempo, Honeycomb, etc.)
    debug=False,                  # Verbose SDK logging
    governance={                  # Opt-in to destructive governance actions
        "allow_stop": False,
        "allow_inject_message": False,
    },
    idle_flush_secs=10,           # How often to flush buffered events
    batch_size=100,               # Max events per /ingest POST
)

---

Skip telemetry for specific calls

Exclude a block from all Syrin instrumentation — useful for health probes and internal calls that shouldn't appear on dashboards.

with syrin_sdk.skip():
    probe = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "ping"}],
    )

---

React to remote config pushes

Register a callback that fires whenever the backend pushes a config update:

@syrin_sdk.on_config_change
def handle_config_change(session_id: str, updates: dict):
    logger.info("Config updated for session %s: %s", session_id, updates)

@syrin_sdk.on_alert
def handle_alert(action: dict):
    if action["level"] == "critical":
        pagerduty.trigger(action["message"])

---

React to every SDK event

Register a callback for any event type emitted by the SDK:

@syrin_sdk.on_event("LLM_CALL")
def on_llm_call(event: dict):
    print(f"Cost: ${event['cost_usd']:.4f}  Model: {event['model']}")

@syrin_sdk.on_event          # no argument → receives every event type
def on_any(event: dict):
    metrics.increment("syrin.events", tags={"type": event["event_type"]})

---

Checkpoints — save and restore conversation state

# Save state before a risky operation
checkpoint = syrin_sdk.create_checkpoint(messages, label="before-tool-call")

try:
    tool_result = call_risky_tool()
    messages.append({"role": "tool", "content": tool_result})
except Exception:
    # Restore to pre-tool state on failure
    messages = checkpoint.messages
    logger.warning("Tool failed — restored to checkpoint %s", checkpoint.checkpoint_id)

---

Multi-instance support

Most apps use the module-level helpers which target the default instance. For processes that need multiple independent SDK instances:

sdk_a = syrin_sdk.init(api_key="...", agent_id="agent-a", instance_name="a")
sdk_b = syrin_sdk.init(api_key="...", agent_id="agent-b", instance_name="b")

# Target a specific instance
sdk_a.configure(temperature=0.3)
sdk_b.configure(temperature=0.9)

---

Environment variables

All init() options can be set via environment variables, which take precedence over code defaults:

Variable	Equivalent init() arg
SYRIN_API_KEY	api_key
SYRIN_BACKEND_URL	backend_url
SYRIN_AGENT_ID	agent_id
SYRIN_DEBUG	debug
SYRIN_CAPTURE_CONTENT	capture_content
SYRIN_OTEL_EXPORTER	otel_exporter
SYRIN_OTEL_ENDPOINT	otel_endpoint
SYRIN_IDLE_FLUSH_SECS	idle_flush_secs
SYRIN_BATCH_SIZE	batch_size

---

API reference

Lifecycle

Symbol	Description
syrin_sdk.init(api_key, ...)	Initialize the SDK, returns SDK instance
syrin_sdk.shutdown()	Flush all pending events and tear down
syrin_sdk.health_check()	Returns True if backend is reachable
syrin_sdk.get_sdk()	Returns the default SDK instance
syrin_sdk.get_session_id()	Returns the active session ID (if inside context())

Scoping

Symbol	Description
syrin_sdk.context(user_id, agent, workflow, swarm, window, ...)	Main context manager — opens a session + scope
syrin_sdk.workflow(id)	Shorthand for context(workflow=id)
syrin_sdk.swarm(id)	Shorthand for context(swarm=id)
syrin_sdk.traced(agent, workflow, ...)	Decorator version of context()
sdk.agent(name)	Returns an AgentHandle with .field(), .run(), .session(), .cfg()

Config

Symbol	Description
syrin_sdk.cfg(key, default, ...)	Declare + read a remotely configurable value
handle.cfg(key, default)	Same, scoped to the agent namespace
sdk.configure(**overrides)	Push local config overrides programmatically

Events & hooks

Symbol	Description
syrin_sdk.log(message, level, metadata)	Emit a custom event on the timeline
syrin_sdk.skip()	Context manager: exclude block from telemetry
syrin_sdk.on_config_change(fn)	Hook: called when backend pushes config
syrin_sdk.on_alert(fn)	Hook: called on backend governance alerts
syrin_sdk.on_event(event_type)(fn)	Hook: called for a specific event type

Governance

Symbol	Description
GovernanceStopError	Raised when backend sends a STOP action
GovernanceStopError.reason	Human-readable reason string
GovernanceStopError.incident_id	Dashboard incident ID
GovernanceStopError.drift_score	Loop/drift score that triggered the stop

Advanced (importable from syrin_sdk.advanced)

Symbol	Description
ConfigGuard	Wrapper that validates, anchors, and rolls back config changes
ConfigFuse	Circuit breaker for repeated config failures
ConfigAnchor	Lock a config key so remote cannot override it
AutoRevert	Automatic rollback to last-good config on crash
tunable / tune()	Decorator + function to mark class fields as remotely tunable
TraceSpan	Manual custom trace spans
SyrinSDKCore	Raw instrumentation engine for framework authors

---

Docs

• 5-minute quickstart
• Remote config, ConfigStore, ConfigGuard
• OpenTelemetry span reference
• Backend API reference
• Adapter system