zespan 1.4.1

Zespan Agent Observability SDK

zespan

LLM observability for Python. Trace every AI call — latency, tokens, cost, errors — with one init().

Install

pip install zespan

Optional extras:

pip install "zespan[otel]"             # OpenTelemetry export
pip install "zespan[llamaindex]"       # LlamaIndex integration
pip install "zespan[haystack]"         # Haystack integration
pip install "zespan[semantic-kernel]"  # Semantic Kernel integration

Quick start

import zespan

zespan.init(
    api_key="zsp_your_key_here",
    project_id="your_project_id",
)
zespan.autopatch()

Get your API key and project ID from app.zespan.com → Project Settings → API Keys.

After init() + autopatch(), all calls to OpenAI, Anthropic, Google GenAI, and other installed providers are automatically traced.

Auto-instrumentation

autopatch() detects and patches every AI provider that is installed:

import zespan
import openai

zespan.init(api_key="zsp_...", project_id="proj_...")
zespan.autopatch()

client = openai.OpenAI()
# This call is automatically traced
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Supported providers:

Function	Provider
patch_openai()	OpenAI
patch_anthropic()	Anthropic
patch_google()	Google Generative AI
patch_openrouter()	OpenRouter
patch_bedrock()	AWS Bedrock
patch_mistral()	Mistral
patch_groq()	Groq
patch_litellm()	LiteLLM

Patch individual providers manually:

from zespan import patch_openai, patch_anthropic

patch_openai()
patch_anthropic()

Configuration

zespan.init(
    api_key="zsp_...",           # required
    project_id="proj_...",       # required — links traces to a project
    environment="production",    # default: "production"
    store_prompts=False,         # store prompt/completion text (default: False)
    redact_keys=["password", "secret", "token", "api_key"],  # keys to redact
    sample_rate=1.0,             # 0.0–1.0, fraction of events to send
    debug=False,                 # log SDK activity to console
    batch_size=50,               # events per batch flush
    flush_interval=2.0,          # seconds between batch flushes
)

User and session context

Tag traces with user ID, session ID, or custom metadata:

from zespan import with_zespan_context

with with_zespan_context(
    user_id="user_123",
    session_id="sess_abc",
    tags={"feature": "chat"},
):
    # All AI calls inside are tagged with this context
    client.chat.completions.create(...)

Works as a decorator too:

@with_zespan_context(user_id="user_123")
def handle_request():
    client.chat.completions.create(...)

Agent tracing

Group multi-step agent workflows into a single trace:

from zespan import with_agent

with with_agent(
    name="SupportAgent",
    role="specialist",
    framework="custom",
    tools=[{"name": "lookup_order", "description": "Look up order by ID"}],
) as agent:
    agent.log_plan(["Look up order", "Check policy", "Draft reply"])

    order = agent.trace_tool(
        "lookup_order",
        {"order_id": "123"},
        lambda: fetch_order("123"),
    )

    agent.delegate_to("RefundAgent", "refund requested")

    # Wrapped LLM calls inside here inherit agent context
    client.chat.completions.create(...)

Agent span types emitted:

span_kind	When
agent	Agent start/end
planning	log_plan()
tool	trace_tool()
handoff	delegate_to()

Manual spans

Trace custom operations that aren't LLM calls:

from zespan import start_span

with start_span(name="document-retrieval", span_kind="retrieval") as span:
    docs = vector_store.search(query)
    span.end(status="success")

Framework integrations

LangChain

from zespan import ZespanCallbackHandler

handler = ZespanCallbackHandler()

chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])

Google ADK

from zespan import ZespanADKTracer

tracer = ZespanADKTracer()
# pass tracer to your ADK runner

Or wrap a specific agent:

from zespan import wrap_adk_agent

traced_agent = wrap_adk_agent(my_agent, name="MyAgent")

LlamaIndex

from zespan import LumiqLlamaIndexCallbackHandler
from llama_index.core import Settings

handler = LumiqLlamaIndexCallbackHandler()
Settings.callback_manager.add_handler(handler)

Haystack

from zespan import LumiqHaystackTracer

tracer = LumiqHaystackTracer()
# use as Haystack tracer

Semantic Kernel

from zespan import instrument_semantic_kernel

instrument_semantic_kernel(kernel)

CrewAI

CrewAI uses OTEL-based instrumentation. Set the env vars Zespan exposes:

from zespan.integrations.crewai_handler import get_crewai_env_vars

env = get_crewai_env_vars()
# apply env to your process before launching CrewAI

AutoGen

from zespan import injectAutoGenContext, extractAutoGenContext

# Inject trace context into an AutoGen message
msg = injectAutoGenContext(msg, trace_id="trace_123")

# Extract it on the receiving end
ctx = extractAutoGenContext(msg)

Prompt management

Fetch versioned prompts from Zespan with 5-minute client-side cache:

import zespan

client = zespan.get_client()

# Latest version
prompt = await client.prompts.get("support-reply")

# Specific version or label
prompt = await client.prompts.get("support-reply", version=3)
prompt = await client.prompts.get("support-reply", label="production")

# Compile variables
text = client.prompts.compile(prompt, {"user_name": "Alice", "product": "Pro"})

Guardrails

Check content against your configured guardrail rules before or after LLM calls:

import zespan

check = zespan.check_guardrails(
    text=user_message,
    phase="pre",
)

if not check["allowed"]:
    return "Request blocked by content policy."

response = client.chat.completions.create(...)

post_check = zespan.check_guardrails(
    text=response.choices[0].message.content,
    phase="post",
)

return post_check["modifiedText"] or response.choices[0].message.content

Flush on exit

The SDK flushes automatically on process exit via atexit. For short-lived scripts or serverless functions, flush manually:

import zespan

zespan.get_client().flush()

Async support

All SDK methods work in async contexts. Context propagation uses Python's contextvars so async tasks inherit their parent's trace context automatically.

import asyncio
import zespan
from zespan import with_zespan_context

zespan.init(api_key="zsp_...", project_id="proj_...")
zespan.autopatch()

async def main():
    with with_zespan_context(user_id="user_123"):
        response = await async_openai_client.chat.completions.create(...)

asyncio.run(main())

OpenTelemetry

Enable native OTEL export alongside Zespan:

zespan.init(
    api_key="zsp_...",
    enable_otel=True,
    otel_endpoint="http://localhost:4318",
    otel_service_name="my-service",
)