ezop 0.0.7

Ezop SDK - The story of every AI agent

Ezop Python SDK

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

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")

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:

# 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:

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:

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.

Parameter	Type	Required	Description
name	str	Yes	Agent name. Together with owner, uniquely identifies the agent on the platform.
owner	str	Yes	Team or user that owns the agent.
version	str	Yes	Version string (e.g. "v1.2.0"). A new version is registered if it does not exist yet.
runtime	str	Yes	Runtime or framework used (e.g. "langchain", "crew", "custom").
description	str	No	Human-readable description of the agent.
default_permissions	list[str]	No	Permissions granted to all versions of this agent by default.
permissions	list[str]	No	Permissions granted to this specific version.
changelog	str	No	Description of what changed in this version.

---

agent.close()

Closes the current run and records its outcome.

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

Parameter	Type	Required	Description
status	str	Yes	Final status of the run. One of "success", "failed", "partial", "canceled", "running".
total_tokens	int	No	Total number of tokens consumed.
total_cost	float	No	Total cost of the run in USD.
message	str	No	Human-readable message describing the outcome, e.g. a failure reason.
metadata	dict	No	Any 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.

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,
)

Parameter	Type	Required	Description
name	str	Yes	Event name (e.g. "llm.call", "tool.invoke").
category	str	Yes	Event category (e.g. "llm", "tool", "retrieval").
span_id	str	No	Identifier to group related events within a run.
status	str	No	Outcome of this event. One of "in_progress", "ok", "error", "cancelled".
input	any	No	Input passed to this step.
output	any	No	Output produced by this step.
metadata	dict	No	Any arbitrary JSON-serialisable data to attach to the event.
error	str	No	Error 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.

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

Parameter	Type	Required	Description
name	str	Yes	Span name (e.g. "llm.call", "tool.invoke").
category	str	Yes	Span category (e.g. "llm", "tool", "retrieval").
input	any	No	Input to record on the opening event.
metadata	dict	No	Any 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:

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