SDK Documentation

Ezop Python SDK

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

Installation Configuration Initialize Track Runs Spans & Events agent.init()agent.close()agent.emit()agent.span()Logging

Installation

>/_ terminal

bash

pip install ezop

Configuration

>/_ terminal

bash

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

Usage

Initialize the agent

>/_ main.py

python

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:

>/_ run.py

python

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

try:
    result = agent.run(user_input)
    agent.close(
        status="success",
        total_tokens=result.usage.total_tokens,
        total_cost=result.cost,
        metadata={"user_id": user_id},
    )
except Exception as e:
    agent.close(status="failed", message=str(e))
    raise

Track steps with spans and events

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

>/_ spans.py

python

# span: emits in_progress on enter, ok/error on exit — same span_id for both
with agent.span("retrieval", category="retrieval", input={"query": user_input}) as s:
    docs = retriever.search(user_input)
    s.set_output({"results": docs})

with agent.span("llm.call", category="llm", input={"prompt": user_input}) as s:
    result = llm.generate(user_input)
    s.set_output(result)

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

agent.close(status="success", total_tokens=result.usage.tokens)

Spans can be nested — child spans automatically record the parent's span_id:

>/_ nested.py

python

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

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

Errors are captured automatically — if an exception is raised inside a span, the closing event is emitted with status="error" and the exception message:

>/_ error.py

python

with agent.span("llm.call", category="llm") as s:
    raise TimeoutError("upstream LLM timeout")
# closing event: status="error", error="upstream LLM timeout"

API Reference

agent.init()

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

ParameterTypeRequiredDescription

namestrYesAgent name. Together with owner, uniquely identifies the agent on the platform.

ownerstrYesTeam or user that owns the agent.

versionstrYesVersion string (e.g. "v1.2.0"). A new version is registered if it does not exist yet.

runtimestrYesRuntime or framework used (e.g. "langchain", "crew", "custom").

descriptionstrNoHuman-readable description of the agent.

default_permissionslist[str]NoPermissions granted to all versions of this agent by default.

permissionslist[str]NoPermissions granted to this specific version.

changelogstrNoDescription of what changed in this version.

agent.close()

Closes the current run and records its outcome.

>/_ close.py

python

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

ParameterTypeRequiredDescription

statusstrYesFinal status of the run. One of "success", "failed", "partial", "canceled", "running".

total_tokensintNoTotal number of tokens consumed.

total_costfloatNoTotal cost of the run in USD.

messagestrNoHuman-readable message describing the outcome, e.g. a failure reason.

metadatadictNoAny arbitrary JSON-serialisable data you want to attach to the run (e.g. user context, request identifiers, feature flags).

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.

>/_ emit.py

python

agent.emit(
    name="llm.call",
    category="llm",
    span_id="span-123",        # optional
    status="success",
    input={"prompt": "hello"},
    output={"text": "hi"},
    metadata={"model": "claude"},
    error=None,
)

ParameterTypeRequiredDescription

namestrYesEvent name (e.g. "llm.call", "tool.invoke").

categorystrYesEvent category (e.g. "llm", "tool", "retrieval").

span_idstrNoIdentifier to group related events within a run.

statusstrNoOutcome of this event. One of "in_progress", "ok", "error", "cancelled".

inputanyNoInput passed to this step.

outputanyNoOutput produced by this step.

metadatadictNoAny arbitrary JSON-serialisable data to attach to the event.

errorstrNoError message if the event failed.

agent.span()

Returns a context manager that tracks a scoped duration. On enter it emits an in_progress event; on exit it emits an ok or error event. Both events share the same span_id so duration can be reconstructed by grouping on span_id and computing the time delta.

>/_ span.py

python

with agent.span("llm.call", category="llm", input={"prompt": prompt}) as s:
    result = llm.generate(prompt)
    s.set_output(result)

ParameterTypeRequiredDescription

namestrYesSpan name (e.g. "llm.call", "tool.invoke").

categorystrYesSpan category (e.g. "llm", "tool", "retrieval").

inputanyNoInput to record on the opening event.

metadatadictNoAny arbitrary JSON-serialisable data to attach to both events.

Call s.set_output(value) inside the block to record the output on the closing event.

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

model.prompt (span_id: A, parent_id: None)

└── tool.call (span_id: B, parent_id: A)

└── memory.read (span_id: C, parent_id: B)

Logging

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

>/_ logging.py

python

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

Log lines are prefixed with ezop and written to stderr by default.