Ezop Python SDK

Ezop tracks the lifecycle of your AI agents — registrations, versions, and runs — so you have full observability across every deployment.

Documentation

Installation

pip install ezop

Configuration

export EZOP_API_KEY=your-ezop-api-key-here
export EZOP_API_URL=https://api.ezop.ai

Usage

Initialize the agent

from ezop import Agent

agent = Agent.init(
    name="customer-support-bot",
    owner="growth-team",
    version="v0.3",
    runtime="langchain",
    description="Handles tier-1 customer support tickets",
    default_permissions=["read:tickets"],
    permissions=["read:tickets", "write:replies"],
    changelog="Switched to new retrieval pipeline",
)

Call Agent.init() once at startup. It is safe to call on every deployment:

• If the agent does not exist, it will be created on the platform.
• If the agent already exists, registration returns the existing agent.
• If a new version is provided, the platform registers it as a new version under the same agent. If the version already exists, version registration is also a no-op.

Track runs

Agent.init() starts a run automatically. Call agent.close() when the invocation is done:

agent = Agent.init(name="my-bot", owner="my-team", version="v1.0", runtime="langchain")

agent.close(
    status="success",
    metadata={"user_id": user_id},
)

Track steps with spans and events

Use span for steps with duration and emit for single points in time:

# span: a record with start_time and end_time
with agent.span("retrieval", metadata={"query": user_input}):
    docs = retriever.search(user_input)

with agent.span("llm.call", metadata={"model": "claude"}):
    result = llm.generate(user_input)

# emit: a single point-in-time event
agent.emit(name="action.selected", category="reasoning")

agent.close(status="success")

Spans can be nested — child spans automatically record the parent's span_id as parent_id, enabling tree reconstruction:

with agent.span("model.prompt") as s1:
    plan = llm.plan(user_input)

    with agent.span("tool.call", metadata={"tool": "stripe.refund"}) as s2:
        refund = stripe.refund(plan.charge_id)
# produces: model.prompt → tool.call (parent_id links them)

Events emitted inside a span are automatically linked to that span via span_id:

with agent.span("llm.call") as s:
    agent.emit(name="token.count", category="llm", metadata={"tokens": 42})
# event.span_id == s.span_id

• Runs — the final outcome of a single agent invocation
• Spans — structured execution steps within a run (e.g. LLM call, retrieval, tool use)
• Events — raw logs inside a span

API Reference

agent.init()

Registers the agent and its version with the Ezop platform, and returns an Agent instance.

agent.close()

Closes the current run and records its outcome.

agent.close(
    status="success",
    message=None,
    metadata={"user_id": "u-123"},
)

agent.emit()

Emits an event on the current run. Events capture discrete steps within a run such as LLM calls, tool invocations, or retrieval operations.

agent.emit(
    name="llm.response",
    category="llm",
    type="llm_response",
    subtype="chain_of_thought",  # optional
    span_id="...",               # optional, auto-set inside a span
    iteration_id=1,              # optional, for loop tracking
    input={"prompt": "hello"},
    output={"text": "hi"},
    metadata={
        "model": "claude-sonnet-4-6",
        "usage": {
            "input_tokens": 12,
            "output_tokens": 8,
        },
    },
    error={"message": "..."},    # optional, jsonb
)

category values and their valid type values:

subtype values:

For LLM events, always include token usage in metadata to enable cost tracking:

agent.emit(
    name="llm.call",
    category="llm",
    metadata={
        "usage": {
            "input_tokens": 12,
            "output_tokens": 8,
        },
    },
)

Popular LLMs return token usage in their responses — map it directly:

# Anthropic
response = anthropic.messages.create(...)
agent.emit(name="llm.call", category="llm", metadata={
    "usage": {
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
    },
})

# OpenAI
response = openai.chat.completions.create(...)
agent.emit(name="llm.call", category="llm", metadata={
    "usage": {
        "input_tokens": response.usage.prompt_tokens,
        "output_tokens": response.usage.completion_tokens,
    },
})

agent.span()

Returns a context manager that tracks a scoped duration as a span record. On enter it creates the span with a start_time; on exit it updates the span with an end_time.

with agent.span("llm.call", metadata={"model": "claude"}) as s:
    result = llm.generate(prompt)

Nested spans automatically propagate context — each child span records the enclosing span's id as its parent_id, enabling tree reconstruction:

model.prompt  (id: A, parent_id: None)
  └── tool.call  (id: B, parent_id: A)
        └── memory.read  (id: C, parent_id: B)

Events emitted inside a span are automatically linked to that span via span_id.

Logging

The SDK uses Python's standard logging module under the ezop namespace. To enable logs in your application:

import logging
logging.getLogger("ezop").setLevel(logging.DEBUG)