Python SDK Reference

`AgentX`

The top-level client. Create one per process.

from agentx import AgentX

client = AgentX(api_key="a1b2c3d4e5f60718293a4b5c6d7e8f9012345678", base_url="http://localhost:4700/api/v1")
# or from environment:
client = AgentX.from_env()   # reads AGENTX_API_KEY, plus the first of
                             # AGENTX_API_BASE_URL / AGENTX_SELFHOST_BASE_URL / BASE_URL

Constructor

Parameter	Type	Default	Description
`api_key`	`str`	`AGENTX_API_KEY` env var	Workspace API key
`base_url`	`str`	`AGENTX_API_BASE_URL` or SDK default	Override API base URL (self-hosted). Include the `/api/v1` suffix, e.g. `http://localhost:4700/api/v1`
`workspace_id`	`str`	`AGENTX_WORKSPACE_ID` env var	Explicit workspace ID, required when the API key belongs to a different user than the workspace being targeted

Class methods

Method	Returns	Description
`AgentX.from_env()`	`AgentX`	Construct from `AGENTX_API_KEY` environment variable

Attributes

Attribute	Type	Description
`client.tracer`	`Tracer`	Tracer and CI/CD evaluation
`client.evaluations`	`EvaluationsRunner`	Dataset-driven evaluation runner, prompts, tool schemas
`client.monitor`	`MonitorClient`	Monitor: patterns, signals, judge scorers, review queue, rules, KPIs/metrics, profiles
`client.outcomes` / `client.feedback`	`OutcomesClient` / `FeedbackClient`	Report real-world outcomes and end-user votes against traces (self-host) - see Outcomes
`client.projects`	`ProjectsClient`	Project CRUD with per-project API keys (self-host)
`client.traces`	`TracesClient`	Read side of tracing: trace by id, paginated listing (self-host)
`client.export`	`ExportClient`	Bulk NDJSON export for backup/migration (self-host) - see Backup & export

`ping()`

client.ping() -> dict

Verifies the client can actually reach AgentX and that the API key is accepted. The constructor is deliberately lazy (no network call) and trace delivery is fire-and-forget, so a wrong base_url or api_key otherwise surfaces only as a one-time log warning while traces silently go nowhere - call this once at startup of a long-running service to fail fast. Raises AgentXConnectionError when the URL is unreachable, AgentXAuthError when the key is rejected, and AgentXAPIError on any other non-OK response. Returns {"ok": True, "base_url": ...} on success.

`close()`

client.close(timeout: float = 5.0) -> bool

Flushes queued traces and stops the tracer’s background ingest worker; returns True when everything drained before the deadline. Optional - an atexit hook already flushes queued traces on interpreter shutdown - but a long-running service that tears clients down mid-process should call it, or use the client as a context manager (with AgentX(...) as client:), so worker threads don’t accumulate.

`Tracer`

Accessed via client.tracer. Handles both tracing and CI/CD evaluation.

`trace()`

tracer.trace(
    name: str,
    *,
    input: Any = None,
    metadata: dict | None = None,
    framework: str | None = None,
    model: str | None = None,
    session_id: str | None = None,
    sync: bool = False,
    monitor: bool | None = None,
    pattern_ids: list[str] | None = None,
    agent_id: str | None = None,
    span_kind: str | None = None,
) -> _TraceSpan

Returns a _TraceSpan that works as a decorator or context manager.

Parameter	Description
`name`	Agent or operation label shown in the UI
`input`	Initial input (context manager only; overrides captured function args)
`metadata`	Arbitrary key-value metadata (not indexed; stored truncated above 100,000 serialized characters - `AGENTX_INGEST_MAX_FIELD_CHARS`)
`framework`	`"langchain"`, `"crewai"`, `"openai-agents"`, `"anthropic"`, or custom
`model`	LLM model name, e.g. `"gpt-4o"`
`session_id`	Groups traces from the same session or conversation thread
`sync`	Sends synchronously (blocking) instead of the default fire-and-forget queue, so `span.trace_id` is populated once the block exits. Works as decorator or context manager - but only the context manager hands you the span to read `trace_id` from. See Linking a trace to an evaluation result.
`monitor`	`False`: opt out of every ingest-time check - pattern/built-in detection, online and custom evaluators, and topics - for traces made inside evaluation runs. `True`: normal checks, plus `pattern_ids` scoping takes effect. `None` (default): normal engine behavior.
`pattern_ids`	With `monitor=True`, restrict detection to exactly these pattern ids (built-ins skipped).
`agent_id`	Pin this trace to a known agent id instead of resolving the agent by `name`.
`span_kind`	Span classification shown in the timeline, e.g. `"llm"`, `"tool"`, `"retrieval"` - any of the full 11-kind vocabulary in Span Kinds.

`flush()`

tracer.flush(timeout: float = 5.0) -> bool

Blocks until all queued traces have been delivered, or until timeout seconds elapse. Returns True when everything drained, False on deadline (a warning is logged and undelivered traces keep sending in the background). Call before process exit in scripts or one-shot jobs.

`use_span()`

tracer.use_span(span: _TraceSpan) -> ContextManager[_TraceSpan]

Makes span (created on another thread) the active span for the duration of the block, on the calling thread. The active-span stack is context-local (a ContextVar): bare threads start with an empty stack, while asyncio tasks inherit a copy of their creator’s. A span opened with tracer.trace(...) on the main thread therefore isn’t automatically visible inside a ThreadPoolExecutor worker (or any other thread). Wrap the worker’s body in use_span() to attach its LLM and tool calls to the parent span instead of starting an independent trace.

with tracer.trace("orchestrator") as span:
    def worker():
        with tracer.use_span(span):
            chain.invoke(query, config={"callbacks": [handler]})

    with ThreadPoolExecutor(max_workers=2) as ex:
        ex.submit(worker).result()

Safe to call concurrently from multiple threads for the same span: each thread pushes and pops on its own stack.

`current_span`

tracer.current_span -> _TraceSpan | None

Property: the innermost with tracer.trace(...) span active in the current context, if any. The manual recorders below attach their child spans to it.

`record_tool_call()`

tracer.record_tool_call(
    name: str,
    *,
    input: Any = None,
    output: Any = None,
    success: bool | None = None,
    error: str | None = None,
    latency_ms: int | None = None,
    start_time: float | None = None,
    end_time: float | None = None,
) -> None

Manually records a tool call that an auto-instrumented framework integration can’t see - e.g. a hand-rolled tool-use loop where the tool executes in plain Python between two model calls. Sent as a tool child span of the active span (see current_span), plus a summary on the root trace’s flat tool_calls list - the surface the engine’s built-in “Tool failure” Monitor check and the dashboard’s Tool quality column read. With no active span, the record is queued onto the next trace this tracer sends (tracer-wide, so wrap in tracer.trace() when concurrent no-span use matters). success=False marks a failed call; leaving success unset means “unknown” and the dashboard falls back to its output-text heuristic.

`trace_tool_call()`

with tracer.trace_tool_call("policy_lookup", input=topic) as t:
    t.output = policy_lookup(topic)

Context manager that times the block and records it via record_tool_call() on exit. An exception escaping the block records the call as failed (success=False plus the error text) and then propagates unchanged.

`record_retrieval()`

tracer.record_retrieval(
    name: str = "Retrieval",
    *,
    query: str | None = None,
    doc_count: int | None = None,
    output: Any = None,
    duration_ms: float | None = None,
    start_time: float | None = None,
    end_time: float | None = None,
    error: str | None = None,
) -> None

Manually records a knowledge-base / vector-store retrieval as a retrieval child span of the active span - the spans the engine reads retrieved chunks from for trajectory-aware judging and the RAG metrics.

`trace_retrieval()`

with tracer.trace_retrieval("kb_search", query=question) as r:
    docs = retrieve(question)
    r.doc_count = len(docs)
    r.output = docs

Context manager that times a retrieval and records it via record_retrieval() on exit. An escaping exception records the retrieval as failed (error set) instead of as a clean empty retrieval, then propagates unchanged.

`record_memory()`

tracer.record_memory(
    name: str = "Memory",
    *,
    operation: str | None = None,
    query: str | None = None,
    output: Any = None,
    duration_ms: float | None = None,
    start_time: float | None = None,
    end_time: float | None = None,
    error: str | None = None,
) -> None

Manually records a long-term-memory operation (a Mem0/Zep/Letta-style recall or store) as a memory child span of the active span. operation is free text - conventionally "read" or "write" - carried in the span’s metadata. record_memory/trace_memory drop when no span is active (warned once per process, then logged at debug), unlike record_tool_call/record_retrieval, which queue onto the next trace.

`trace_memory()`

with tracer.trace_memory("user prefs", operation="read", query=user_id) as m:
    m.output = memory.search(user_id, question)

Context manager that times a memory operation and records it via record_memory() on exit. An escaping exception records the operation as failed, then propagates unchanged.

`run_eval()`

tracer.run_eval(
    dataset_id: str,
    agent_fn: Callable[[str], str],
    *,
    agent_name: str | None = None,
    pass_rate_threshold: float | None = None,
    git_context: dict | None = None,
    concurrency: int = 1,
    fail_on_gate: bool = False,
    timeout_per_question: float | None = None,
) -> CIRunResult

High-level CI/CD evaluation: creates a run, calls agent_fn for every test case, submits results, finalizes, and returns the gate decision. Raises CIGateFailure (containing the CIRunResult) when fail_on_gate=True and the gate is "fail".

`create_ci_run()`

tracer.create_ci_run(
    dataset_id: str,
    *,
    agent_name: str | None = None,
    pass_rate_threshold: float | None = None,
    git_context: dict | None = None,
    workspace_id: str | None = None,
) -> CIRun

`submit_result()`

tracer.submit_result(
    run_id: str,
    question_index: int,
    output: Any,
    *,
    input: Any = None,
    latency_ms: int | None = None,
) -> CIQuestionScore

`finalize_ci_run()`

tracer.finalize_ci_run(run_id: str) -> CIRunResult

`get_ci_run()`

tracer.get_ci_run(run_id: str) -> CIRunStatus

`evaluate_trace()`

tracer.evaluate_trace(
    trace_id: str,
    dataset_id: str,
    *,
    question_index: int | None = None,
) -> dict

Score a previously-recorded trace against a dataset. The agent is not re-run. Returns { run_id, trace_id, rating, justification, status }. trace_id comes from a prior tracer.trace(..., sync=True) call’s span.trace_id.

`_TraceSpan`

Returned by tracer.trace(). Can be used as a decorator or context manager.

Attributes

Attribute	Type	Description
`span.input`	`Any`	Agent input. Assign in context manager to override captured args.
`span.output`	`Any`	Agent output. Must be set manually in context manager mode.
`span.tool_calls`	`list`	Accumulated tool calls. Append with `add_tool_call()`.
`span.trace_id`	`str \| None`	The ingested trace’s id, readable once the block exits. Only populated when opened with `sync=True`; `None` otherwise, since the default mode queues the send and never learns the id.

Methods

`add_tool_call()`

span.add_tool_call(
    name: str,
    *,
    input: Any = None,
    output: Any = None,
    latency_ms: int | None = None,
) -> None

Records a tool call made during this span. Safe to call multiple times.

`set_error()`

span.set_error(message: str) -> None

Marks this span as failed. Takes precedence over any exception automatically caught by __exit__.

Usage patterns

@tracer.trace("my-agent", framework="langchain", model="gpt-4o")
def run(query: str) -> str:
    return chain.invoke(query)
# input = captured from function args
# output = return value

with tracer.trace("my-agent") as span:
    span.input = {"query": query, "user_id": user_id}
    kb = search_kb(query)
    span.add_tool_call("search_kb", input=query, output=kb, latency_ms=180)
    span.output = llm.invoke(f"Context: {kb}\n\n{query}")

with tracer.trace("my-agent", sync=True) as span:
    span.output = call_agent(query)

print(span.trace_id)  # None unless sync=True

@tracer.trace("async-agent", framework="openai-agents")
async def run(query: str) -> str:
    result = await runner.run(agent, query)
    return result.final_output

with tracer.trace("my-agent") as span:
    try:
        span.output = call_agent(query)
    except Exception as e:
        span.set_error(str(e))
        raise

CI/CD Dataclasses

`CIRun`

Returned by create_ci_run().

@dataclass
class CIRun:
    run_id: str
    dataset_id: str
    total_questions: int
    test_cases: list[CITestCase]
    expires_at: str          # ISO 8601, run expires if not finalized within 2 hours

`CITestCase`

@dataclass
class CITestCase:
    index: int
    query: str | None        # None when ci.exposeTestInputs is False

`CIQuestionScore`

Returned by submit_result().

@dataclass
class CIQuestionScore:
    question_index: int
    rating: int              # 0-10
    justification: str
    passed: bool
    gate_fired: bool         # True when failFast fired, stop submitting
    input: Any
    output: Any

`CIRunResult`

Returned by finalize_ci_run() and run_eval().

@dataclass
class CIRunResult:
    run_id: str
    gate: Literal["pass", "fail"]
    pass_rate: float                      # 0.0 - 1.0
    total_questions: int
    passed_questions: int
    scores: list[CIQuestionScore]
    violations: list[ThresholdViolation]
    git_context: dict | None
    finalized_at: str | None

`CIRunStatus`

Returned by get_ci_run().

@dataclass
class CIRunStatus:
    run_id: str
    status: Literal["in_progress", "completed", "failed"]
    gate: Literal["pass", "fail"] | None  # None while in_progress
    results_submitted: int
    total_questions: int
    created_at: str
    expires_at: str
    finalized_at: str | None
    git_context: dict | None

`ThresholdViolation`

@dataclass
class ThresholdViolation:
    question_index: int
    metric: str              # e.g. "rating"
    threshold: float
    actual: float
    question_text: str

Exceptions

All exceptions inherit from AgentXError.

from agentx import (
    AgentXValidationError,
    AgentXConnectionError,
    AgentXError,
    AgentXAuthError,
    AgentXAPIError,
    DatasetNotFound,
    CINotEnabled,
    CIRunExpired,
    CIGateFailure,
)

Exception	When raised
`AgentXError`	Base class for all SDK errors
`AgentXAuthError`	Invalid or missing API key
`AgentXConnectionError`	The API (or self-host engine) could not be reached at the configured `base_url`, e.g. from `client.ping()`.
`AgentXValidationError`	Request arguments the engine rejected before processing (HTTP 422)
`AgentXAPIError`	Unexpected API error. Has `.status_code: int \| None` attribute.
`DatasetNotFound`	Dataset ID does not exist or is not accessible
`CINotEnabled`	Dataset exists but `ci.enabled` is `False`
`CIRunExpired`	CI run was not finalized within 2 hours
`CIGateFailure`	Gate is `"fail"` and `fail_on_gate=True`. Has `.result: CIRunResult` attribute.

`CIGateFailure`

try:
    result = client.tracer.run_eval(
        dataset_id=dataset_id,
        agent_fn=my_agent,
        fail_on_gate=True,
    )
except CIGateFailure as e:
    print(f"Pass rate: {e.result.pass_rate:.0%}")
    for v in e.result.violations:
        print(f"  Q{v.question_index}: {v.metric} = {v.actual:.2f} (threshold {v.threshold:.2f})")
    sys.exit(1)

Environment variables

Variable	Description
`AGENTX_API_KEY`	Workspace API key (required)
`AGENTX_API_BASE_URL`	Override API base URL for self-hosted deployments
`AGENTX_SELFHOST_BASE_URL`	Fallback for the base URL when `AGENTX_API_BASE_URL` is unset - matches the self-host samples’ .env convention
`BASE_URL`	Last-resort fallback for the base URL, checked after the two above
`AGENTX_WORKSPACE_ID`	Explicit workspace ID, injected into every request automatically

​AgentX

​Constructor

​Class methods

​Attributes

​ping()

​close()

​Tracer

​trace()

​flush()

​use_span()

​current_span

​record_tool_call()

​trace_tool_call()

​record_retrieval()

​trace_retrieval()

​record_memory()

​trace_memory()

​run_eval()

​create_ci_run()

​submit_result()

​finalize_ci_run()

​get_ci_run()

​evaluate_trace()

​_TraceSpan

​Attributes

​Methods

​add_tool_call()

​set_error()

​Usage patterns

​CI/CD Dataclasses

​CIRun

​CITestCase

​CIQuestionScore

​CIRunResult

​CIRunStatus

​ThresholdViolation

​Exceptions

​CIGateFailure

​Environment variables

`AgentX`

Constructor

Class methods

Attributes

`ping()`

`close()`

`Tracer`

`trace()`

`flush()`

`use_span()`

`current_span`

`record_tool_call()`

`trace_tool_call()`

`record_retrieval()`

`trace_retrieval()`

`record_memory()`

`trace_memory()`

`run_eval()`

`create_ci_run()`

`submit_result()`

`finalize_ci_run()`

`get_ci_run()`

`evaluate_trace()`

`_TraceSpan`

Attributes

Methods

`add_tool_call()`

`set_error()`

Usage patterns

CI/CD Dataclasses

`CIRun`

`CITestCase`

`CIQuestionScore`

`CIRunResult`

`CIRunStatus`

`ThresholdViolation`

Exceptions

`CIGateFailure`

Environment variables