Python SDK reference

acruxcore is the async Python SDK for AcruxCore. It renders stored prompts, calls the gateway (with streaming, tools, and structured output), runs the tool-calling loop, reports and reads traces, and manages the tool catalog. Every method below has a 1:1 Node counterpart in @acruxcoreai/sdk — only the casing and a few option shapes differ.

pip install acruxcore

AcruxCore is async, so it is an async context manager. For code that is not async, use AcruxCoreSync, which has the same methods. Create one instance at startup and reuse it.

Construct the client

import asyncio
from acruxcore import AcruxCore

async def main():
    # Reads ACRUXCORE_API_KEY and ACRUXCORE_BASE_URL from the environment.
    async with AcruxCore() as hub:
        ...

asyncio.run(main())

Parameter	Type	Default	Notes
api_key	str	ACRUXCORE_API_KEY	Raises MISSING_API_KEY if neither is set.
base_url	str	ACRUXCORE_BASE_URL	Raises MISSING_BASE_URL if neither is set.
cache_ttl	int (ms)	60000	Render cache window. 0 disables caching (and serve-stale).
max_cache_size	int	500	Max LRU entries. Set by the first instance.
max_retries	int	1	Retries this SDK's own request to AcruxCore on a transient failure (2 total attempts). It does not change how many times a provider is called — that is the per-call gateway argument.
retry_interval	int (ms)	500	Delay between retries.
timeout	float (s)	30.0	Per-request timeout.
transport	httpx.AsyncBaseTransport	—	For testing/injection.
provider	ProviderConfig	—	Client-level BYO default; overridden by a per-call provider.

Errors raise AcruxCoreError with a machine-readable code — see Error codes.

Sync client: AcruxCoreSync

For code that is not async — a script, a Flask or Django app. It takes the same arguments and has the same namespaces, but every method blocks and returns the result. A streaming call returns a plain iterator.

from acruxcore import AcruxCoreSync

with AcruxCoreSync() as hub:
    reply = hub.gateway.chat("support-model", messages)
    for chunk in hub.gateway.stream("support-model", messages):
        print(chunk.delta.get("content", ""), end="")

One client can be shared across threads. It runs the async client on its own background thread. close() (or leaving the with block) sends buffered traces; so does interpreter exit, or the client being garbage-collected. A stream you leave with break is closed when the iterator is dropped; call its close() to release it at once. Do not call it from a sync tool function that run_tool_loop runs: that raises VALIDATION_ERROR, because the call would wait forever. Make that tool async and use AcruxCore inside it. Not in 0.17.0; comes in the next release.

prompts.render(name, alias, variables=None)

Render a stored prompt by name + alias into templated messages, plus the tools attached to that version. Cached per (api_key, name, alias, variables) with stale-while-revalidate: a fresh hit returns immediately, a stale hit returns immediately and refreshes in the background, a cold miss fetches.

rendered = await hub.prompts.render(
    "support-reply", "production",
    {"company": "Acme", "customer_message": "Order #123 is late"},
)
print(rendered.messages, rendered.model)

Parameter	Type	Required	Notes
name	str	yes	Prompt slug (not id).
alias	str	yes	e.g. "production", "staging".
variables	dict	no	Template variables. Defaults to {}.

Returns RenderResult(messages, tools, model, version_id, version_number, variables). model is the version's bound default (or None); variables echoes back what you called with. Pass version_id as prompt_version_id and variables as variables to gateway.chat()/gateway.run_tool_loop() — the pair is what gives a trace prompt lineage and lets feedback on it become an evaluation dataset example. gateway.run_prompt_with_tools(rendered) does both for you. Raises MISSING_VARIABLES if the template needs a variable you did not supply.

Prompt lifecycle (hub.prompts)

prompts.list(*, search=None, page=None, limit=None)

List prompts for the team, newest first.

page = await hub.prompts.list(search="support", limit=10)

Parameter	Type	Notes
search	str	Free-text search on name.
page	int	1-based page.
limit	int	Page size.

Returns PromptListResult(data, total, page, limit).

prompts.get(prompt_id)

Fetch one prompt by id.

prompt = await hub.prompts.get(prompt_id)

Returns PromptDetail (id, name, description, version_count, created_at, updated_at).

prompts.create(name, *, description=None)

Create a new prompt shell. Commit a version with commit_version to give it content.

prompt = await hub.prompts.create("support-bot", description="Customer support")

Returns PromptDetail.

prompts.update(prompt_id, **kwargs)

Update a prompt's name and/or description. Does not touch versions.

updated = await hub.prompts.update(prompt_id, description="v2 description")

Returns PromptDetail.

prompts.delete(prompt_id)

Delete a prompt and every version/alias under it. Returns None.

await hub.prompts.delete(prompt_id)

prompts.commit_version(prompt_id, messages, *, model=None, tools=None)

Commit a new immutable version for a prompt. The first commit auto-creates production and staging aliases.

version = await hub.prompts.commit_version(
    prompt_id,
    messages=[
        {"role": "system", "content": "You are a support agent."},
        {"role": "user", "content": "{{question}}"},
    ],
    model="gpt-4o-mini",
)

Returns VersionDetail (id, version_number, messages, model, tools, aliases). aliases is present only on the first version.

prompts.list_versions(prompt_id, *, page=None, limit=None)

List a prompt's versions, newest first. Items omit messages — use get_version for full content.

versions = await hub.prompts.list_versions(prompt_id)

Returns VersionListResult(data, total, page, limit).

prompts.get_version(prompt_id, version_number)

Fetch one version with its full messages.

v = await hub.prompts.get_version(prompt_id, 1)

Returns VersionDetail.

prompts.get_by_alias(name, alias)

The version an alias points at, found by prompt name. Its messages are the template, with {{ placeholders }} still in them. Use it when your app renders templates itself; use prompts.render when the server should fill them in. Pass version.id as prompt_version_id to link the model call to this version.

version = await hub.prompts.get_by_alias("support-reply", "production")
messages = my_render(version.messages, ticket=ticket)
reply = await hub.gateway.chat("support-model", messages, prompt_version_id=version.id)

Returns VersionDetail, a fresh copy on every call, so changing its messages in place is safe. Cached for cache_ttl per (name, alias), up to max_cache_size entries; a miss makes one request (three against a server older than GET /prompt-versions). Raises API_ERROR with status 404 when no prompt has that exact name, or the prompt has no such alias. Not in 0.17.0; comes in the next release.

prompts.diff(prompt_id, from_version, to_version)

Diff two versions.

diff = await hub.prompts.diff(prompt_id, 1, 2)

Returns DiffResult (changes).

prompts.promote_alias(prompt_id, alias, version_number)

Point an alias (e.g. "production") at a specific version. Creates the alias if it doesn't exist.

alias = await hub.prompts.promote_alias(prompt_id, "production", 3)

Returns AliasDetail (alias, version_number, prompt_id).

prompts.list_aliases(prompt_id)

Read every alias on a prompt and the version each one points at — the read half of promote_alias, for a deploy check or a CI guard asking "which version is production on right now?".

for a in await hub.prompts.list_aliases(prompt_id):
    print(a.alias, "->", f"v{a.version_number}")

Returns list[AliasDetail]. Empty for a prompt with no committed version, since the first version is what mints the aliases. Added in 0.10.0.

prompts.export_version(prompt_id, version_number)

Export a version for portability (JSON blob).

exported = await hub.prompts.export_version(prompt_id, 1)

Returns ExportedPromptVersion.

prompts.import_prompt(export_data)

Import an exported prompt as a new prompt with one version.

imported = await hub.prompts.import_prompt(exported)

Returns ImportPromptResult (prompt_id, prompt_name, version_number).

prompts.traces_for_version(prompt_id, version_number, *, limit=None, page=None)

List traces that used a specific prompt version.

traces = await hub.prompts.traces_for_version(prompt_id, 1, limit=10)

Returns TraceListResult(data, total, page, limit).

Which call runs your tools

You have	Call
Messages you wrote, no tools	gateway.chat
Messages you wrote, and the tools named on the call	gateway.run_tool_loop
A prompt from the catalog	prompts.render, then gateway.run_prompt_with_tools

chat() takes tools too, but only offers them: it returns with finish_reason="tool_calls" and the request on message["tool_calls"], and running it is yours to do. It also rejects a function decorated with @acrux.tool — that shape belongs to run_tool_loop, and the error says so.

Inside the loop, what you pass depends on where the definition lives and whose process runs the call:

Definition lives in	Runs in	Pass
Your code, declared with @acrux.tool	Your process	tools=[get_weather]
The catalog, http executor	AcruxCore	Nothing — the prompt's binding is enough, or tool_refs=[{"name": ...}]
The catalog, client executor	Your process	client_tools={"get_weather": get_weather}
Nowhere — inline for this one call	Your process	tool_defs=[...] and dispatch=

dispatch is the most general option. Use dispatch when the tool names are not known until runtime.

Call a prompt's tools from the SDK works through each shape end to end, streaming included.

gateway.chat(...)

One gateway completion at POST /gateway/chat/completions — no tool-dispatch loop. If the model returns tool_calls, they are handed back raw on result.message["tool_calls"]; use gateway.run_tool_loop to dispatch them.

reply = await hub.gateway.chat("support-model", rendered.messages, temperature=0.2)
print(reply.content, reply.usage)

For streaming, use gateway.stream() which returns an async iterator of chunks:

stream = await hub.gateway.stream("support-model", rendered.messages)
async for chunk in stream:
    print(chunk.delta.get("content", ""), end="", flush=True)

To stop reading early, break and then await stream.aclose(), which closes the connection. A BYO stream closed this way still records its llm span, with the text received so far and the attribute clientClosed: true. Token usage is usually missing on that span, since the provider sends it last.

Parameter	Type	Required	Notes
model	str	yes	Model public name.
messages	list[dict]	yes	Chat messages.
tools	list[ToolDefinition]	no	Inline OpenAI-shaped tool definitions.
tool_refs	list[{name, alias?}]	no	Catalog tool references (resolved server-side).
tool_choice	ToolChoice	no	How the model uses tools.
response_format	ResponseFormat	no	Structured output. Mutually exclusive with tools/tool_choice.
temperature	float	no	Sampling temperature.
max_tokens	int	no	Max completion tokens.
reasoning_effort	str	no	How hard a reasoning model thinks, e.g. "low" or "high". Sent as reasoning_effort. Not in 0.17.0; comes in the next release.
stream	bool	no	Return an async iterator of ChatChunk instead of ChatResult.
provider	ProviderConfig	no	Per-call BYO override.
gateway	{"max_retries": int, "fallback": bool}	no	Per-call gateway controls: how many times the provider may be retried (0–5), and whether a failure may fall back to the next model in the chain. A different layer from the client's max_retries, which only retries this SDK's request to AcruxCore. Ignored on a BYO call.
prompt_version_id	str	no	From prompts.render().version_id; stamped on the trace span.
variables	dict	no	From prompts.render().variables. Recorded on the span, never sent to a BYO provider. Send it whenever prompt_version_id is set — without it the run cannot seed a dataset. With no prompt_version_id, a gateway call renders {{ placeholders }} in your messages with these instead.
trace	bool | dict	no	False reports nothing. A dict takes name, trace_id, parent_span_id, session_id, tags and metadata. parent_span_id puts this call's llm span under a span you reported. Inside a step, the call joins the step's trace on its own.
extra_body	dict	no	More OpenAI fields (top_p, seed, stop, …) added to the request body. Raises VALIDATION_ERROR for a key a named argument already sets, or for model, messages or stream. The gateway forwards only fields it knows and names the rest in its x-gateway-params-ignored response header; a BYO provider gets them all. Not in 0.17.0; comes in the next release.

Returns ChatResult (id, model, content, message, finish_reason, usage, gateway) or, when streaming via gateway.stream(), an async iterator of ChatChunk. gateway carries request_id, provider, model, cost_usd, cache, trace_id, and span_ref read from the gateway's x-gateway-* headers. usage also carries cached_tokens (part of prompt_tokens) and reasoning_tokens (part of completion_tokens) when the provider reports them.

note

response_format and tools/tool_choice/tool_refs cannot ride the same gateway request — the gateway returns a 400. To get a typed answer that also calls tools, pass both to gateway.run_tool_loop; the SDK gathers with tools, then makes one shaping call with the format.

gateway.run_tool_loop(...)

The full agent loop: calls the model, runs the tools it asks for, appends the results, and repeats until the model stops calling tools or max_iterations is hit. Tools requested in one turn run concurrently.

from acruxcore import acrux

@acrux.tool
async def get_weather(city: str) -> dict:
    """Get the current weather for a city.

    Args:
        city: City name, e.g. 'Lahore'.
    """
    return {"tempC": 21}

result = await hub.gateway.run_tool_loop(
    "agent-model",
    [{"role": "user", "content": "Weather in Lahore?"}],
    tools=[get_weather],
)
print(result.content, result.trace_id)

Parameter	Type	Required	Notes
model	str	yes	Model public name.
messages	list[dict]	yes	Seed messages.
tools	list[Callable]	no	Functions decorated with @acrux.tool. Run locally; reconciled with the catalog.
tool_defs	list[ToolDefinition]	no	Raw OpenAI definitions, sent inline; route to dispatch.
tool_refs	list[{name, alias?}]	no	Catalog refs. http executor runs on the platform.
client_tools	dict[str, Callable]	no	Catalog tool name → the function that runs it, for client executors. Writes nothing to the catalog; keeps the binding's alias or pin. Called with the schema's own field names as keywords.
dispatch	Callable[[str, dict], Any]	no	(name, args) -> result (sync or async). Required for tool_defs, and the fallback for a client ref with no decorated tool and no client_tools entry.
sync	bool	no	Reconcile tools with the catalog first. Default True.
max_iterations	int	no	Max round-trips. Default 10.
temperature	float	no	Sampling temperature.
max_tokens	int	no	Max completion tokens.
reasoning_effort	str	no	How hard a reasoning model thinks, e.g. "low" or "high". Sent as reasoning_effort. Not in 0.17.0; comes in the next release.
response_format	ResponseFormat	no	Shapes the final answer; may be combined with tools (gather + shape).
trace	bool | {trace_id?, name?, session_id?}	no	Default True.
provider	ProviderConfig	no	Per-call BYO override.
gateway	{"max_retries": int, "fallback": bool}	no	Per-call gateway controls: how many times the provider may be retried (0–5), and whether a failure may fall back to the next model in the chain. A different layer from the client's max_retries, which only retries this SDK's request to AcruxCore. Ignored on a BYO call. Applied to every round of the loop.
prompt_version_id	str	no	Stamped on every llm span this loop records.
variables	dict	no	Stamped alongside it. Same argument as on chat().

Returns RunToolLoopResult(content, messages, iterations, stopped_at_limit, trace_id). Raises MISSING_DISPATCH before the first model call if a tool has no runner — the message names the tool and, when client_tools was passed, lists the keys it held. Raises VALIDATION_ERROR, also before the first call, when a client_tools function cannot receive its tool's required arguments.

gateway.run_prompt_with_tools(rendered, ...)

The same loop, with everything it needs taken from a render result: the version's bound model, the rendered messages, the tools bound to this prompt alias, and the version id that stamps trace lineage. Your code adds the user's turn and the tools it runs itself.

rendered = await hub.prompts.render("travel-planner", "production", {"today": "2026-08-21"})
messages = [*rendered.messages, {"role": "user", "content": question}]

result = await hub.gateway.run_prompt_with_tools(
    rendered,
    messages=messages,
    client_tools={"search_flights": search_flights},
)

Parameter	Type	Required	Notes
rendered	RenderResult	yes	From prompts.render. Positional.
model	str	no	Overrides the version's bound model.
messages	list[dict]	no	Overrides the rendered messages.
tool_refs	list[{name, alias?}]	no	Overrides the prompt's bindings entirely. [] runs the prompt with no tools.
client_tools	dict[str, Callable]	no	The prompt's client-executor tools, keyed by name. Its http tools need no entry — the platform runs them.
stream	bool	no	True returns the same event stream as run_tool_loop.

Every other keyword of run_tool_loop is accepted and passes straight through. A prompt with no tools bound still runs, as a plain completion.

Returns the same RunToolLoopResult, or an event stream when stream=True. Raises VALIDATION_ERROR when the version has no bound model and no model= was passed.

traces.ingest(input, *, wait=True)

Report a trace (a group of spans) to AcruxCore. A single-trace convenience over the batch endpoint — omit traceId to mint a new trace, pass one to append.

result = await hub.traces.ingest({
    "name": "rag-pipeline",
    "spans": [
        {"spanId": "retrieval-1", "name": "vector-search", "kind": "retrieval",
         "status": "ok", "startTime": "2026-01-01T00:00:00Z",
         "endTime": "2026-01-01T00:00:01Z",
         "input": {"query": "shipping policy"}, "output": {"hits": 4}},
    ],
})
print(result.trace_id)

Field	Type	Required	Notes
traceId	str	no	Omit to mint a new trace; pass one to append.
sessionId	str	no	Groups traces into a session.
name	str	no	Trace name.
capturePayloads	bool	no	Force payload capture on for this trace.
tags	list[str]	no	Set on creation; merged (union) on append.
metadata	dict	no	Set on creation; shallow-merged on append.
spans	list[IngestSpan]	yes	The spans to report — see Span shapes.

Returns TraceResult(trace_id).

Reporting without waiting

wait=True (the default) awaits the POST: errors are raised at the call site and trace_id is the server's. wait=False buffers the trace and returns immediately, so instrumenting a retrieval or rerank step costs no round trip:

result = await hub.traces.ingest({"name": "retrieval", "spans": [...]}, wait=False)

# Usable straight away — the id is generated client-side, and the API creates
# the trace under it. Hand it to the gateway so both land on one trace.
await hub.gateway.chat(model="gpt-4o-mini", messages=[...], trace={"trace_id": result.trace_id})

await hub.traces.flush()   # before reading it back, or before the process exits

The trade: nothing is confirmed at the call site. A failed send warns once per error kind and drops the batch, the same way the gateway's own span reporting already behaves. traces.flush() waits for everything buffered — including the gateway's spans, which share the buffer — and AcruxCore.aclose() flushes too.

traces.flush()

Wait for every trace buffered by ingest(..., wait=False) (and by the gateway's own span reporting) to be sent. Returns once the queue is empty.

traces.step(name, *, kind="chain", ...)

Trace one step of your own code — a retrieval, a parse, a whole request handler. Use it as a decorator on a sync or async function, or as a with / async with block. On a generator function, the step stays open until the generator ends, and calls made by the code that reads it are not counted as part of the step.

@hub.traces.step("classify", kind="chain")
async def classify(ticket: str) -> str:
    reply = await hub.gateway.chat("support-model", [{"role": "user", "content": ticket}])
    return reply.content

@hub.traces.step("handle-ticket", kind="agent")
async def handle(ticket: str) -> str:
    return await classify(ticket)

This gives one trace named handle-ticket, with classify under it and the model call under classify:

handle-ticket   agent
└─ classify     chain
   └─ llm       support-model

A step opened inside another step is its child. A gateway.chat() or gateway.stream() inside a step joins the step's trace, with its llm span under the step. This works on the gateway path and the BYO path. run_tool_loop does not join on its own; pass it trace={"trace_id": step.trace_id}, and the trace still takes the step's name.

Parameter	Type	Default	Notes
name	str	—	The span's name. The outermost step's name is also the trace's name.
kind	str	"chain"	llm, tool, retrieval, embedding, agent, chain or other.
input	Any	the arguments	A decorated function records its arguments by name (without self).
capture_input / capture_output	bool	True	False records no input / output.
session_id	str	—	Session for the trace.
tags / metadata	list / dict	—	Added to the trace, from any step.

The return value is the step's output. In a with block, set it yourself:

async with hub.traces.step("retrieve", kind="retrieval") as step:
    docs = await search(query)
    step.output = {"hits": len(docs)}

step.span_id and step.trace_id are readable inside the block. A value that is not JSON is stored as its repr. An exception marks the span status="error" and is raised again unchanged. One step object can be entered again, or by several tasks at once; each entry records its own output.

A step only collects calls made on the client that opened it. A call on a second client inside the step starts that client's own trace, because the two clients may belong to different teams. One trace holds at most 1,000 spans. Spans past that are dropped, and the root span's droppedSpans attribute counts them.

The whole tree of the outermost step is sent as one report when that step ends, in the background. So the trace appears once the request finishes, not span by span. A step that is still running when the outermost step ends — a task it started without awaiting — is sent when it ends. Call traces.flush() before reading it back. Not in 0.17.0; comes in the next release.

traces.submit_feedback(...) / traces.update_feedback(...)

Attach feedback to a trace (or one span), then edit it in place. Only the original author may edit.

fb = await hub.traces.submit_feedback(trace_id, rating=1, label="helpful", comment="Resolved my issue.")
await hub.traces.update_feedback(trace_id, fb.id, rating=1, label="unhelpful")

submit_feedback(trace_id, *, ...) — at least one of rating/label/comment:

Field	Type	Notes
trace_id	str	Required (positional).
span_id	str	Scope to one span.
rating	int	1 (thumbs up) or -1 (thumbs down).
label	str	Short label.
comment	str	Free text.
source	str	"user" | "developer" | "end_user" | "api".

update_feedback(trace_id, feedback_id, *, rating=..., label=..., comment=...) — pass a value to change it, None to clear it, or omit the argument to keep the existing value (uses an Ellipsis sentinel internally).

Returns FeedbackResult (id, trace_id, span_id, rating, label, comment, source, created_by, created_at, updated_at).

traces.get(trace_id) / traces.list(...)

Read traces back. traces.get returns the header plus the full span tree; traces.list returns one page of summaries, newest first.

full = await hub.traces.get(trace_id)
print(full.trace, full.spans)

page = await hub.traces.list(status="error", min_latency_ms=2000, limit=20)
print(page.data, page.total)

traces.list(*, ...) filters — all keyword-only and optional:

Field	Type	Notes
from_ / to	str	ISO date range (from_ maps to from).
status	str	'ok' | 'error' | 'unset'.
model	str	Filter by model.
session_id	str	Filter by session.
prompt_version_id	str	Filter by prompt version.
min_latency_ms	int	Minimum latency.
min_cost_usd	float	Minimum cost.
min_tokens	int	Minimum total tokens.
q	str	Free-text search.
error_type	SpanErrorType	Traces with a span that failed this way, e.g. 'tool_declared'. Not in 0.19.0; comes in the next release.
error_code	str	Traces with a span carrying this declared error code. Not in 0.19.0; comes in the next release.
has_warning	bool	True for traces with a warning, False for none. Not in 0.19.0; comes in the next release.
tags	Sequence[str]	Traces carrying all of these tags. Not in 0.19.0; comes in the next release.
metadata	Mapping[str, str]	Traces whose metadata has every one of these pairs. Not in 0.19.0; comes in the next release.
page / limit	int	Pagination (1-based page).

traces.get returns GetTraceResult(trace, spans); traces.list returns ListTracesResult(data, total, page, limit).

Trace analytics (hub.traces)

traces.analytics(*, group_by=None, since=None, until=None)

Aggregated trace metrics (latency, cost, tokens) grouped by a facet.

analytics = await hub.traces.analytics(group_by="model")

Parameter	Type	Notes
group_by	str	Facet key to group by (e.g. "model", "status").
since / until	str	ISO-8601 time bounds.

Returns AnalyticsResult (data).

traces.list_facets()

Discover available facet keys for grouping.

facets = await hub.traces.list_facets()

Returns TraceFacets — object whose keys are facet names.

traces.get_facet_values(key)

Get distinct values for a facet key.

values = await hub.traces.get_facet_values("model")

Returns FacetValuesResult (values).

traces.get_settings() / traces.update_settings(capture_payloads)

Read or update the team's trace capture settings.

settings = await hub.traces.get_settings()
await hub.traces.update_settings(True)  # enable payload capture

Returns TraceSettings (capture_payloads).

traces.get_feedback_summary(*, since=None, until=None)

Aggregated feedback buckets (rating distribution).

summary = await hub.traces.get_feedback_summary()

Returns FeedbackSummaryResult (data).

traces.list_feedback(*, limit=None, page=None)

Paginated feedback list across all traces.

feedback = await hub.traces.list_feedback(limit=10)

Returns FeedbackListResult (data, total, page, limit).

traces.get_trace_feedback(trace_id)

All feedback for a specific trace.

fb = await hub.traces.get_trace_feedback(trace_id)

Returns TraceFeedbackResult (data).

gateway.flush() / gateway.aclose() / aclose()

await hub.gateway.flush()    # wait for background trace writes to finish
await hub.gateway.aclose()   # flush, then stop accepting spans
await hub.aclose()           # the above, then close the HTTP connection pool

gateway.chat(), streaming, and gateway.run_tool_loop() hand back their result without waiting for the trace write — call gateway.flush() before reading the traces API back. A script that returns from main() does not need either: the SDK drains at interpreter exit.

hub.aclose() is the full shutdown, and it is what async with AcruxCore() as hub: calls on exit. Reach for it directly wherever async with is not available — a notebook cell, a REPL, or a server's own shutdown hook. gateway.aclose() stops at the span queue and leaves the connection pool open, so code that builds a client per request or per job should call hub.aclose() instead.

Tool catalog (hub.tools)

Catalog operations live on hub.tools.

tools.sync(...) / tools.sync_one(...)

Reconcile tools from @acrux.tool against the catalog. Idempotent and cached per process on the spec hash — a second call with an unchanged tool makes no request.

await hub.tools.sync([get_weather])
one = await hub.tools.sync_one(get_weather.__acrux_tool__, on_conflict="error")

Parameter	Type	Notes
tools / spec	list[Callable] / ToolSpec	Decorated functions (for sync) or a ToolSpec (for sync_one).
on_conflict	str	Default "warn". "error" raises when a commit supersedes a dashboard-authored version.

Returns a list of ToolSyncResult (tool_id, version_number, committed, alias, superseded_source) from sync, or a single one from sync_one. committed is False on a cache hit.

tools.resolve(refs)

Resolve catalog refs to schemas plus executor types in one request.

[resolved] = await hub.tools.resolve([{"name": "get_weather", "alias": "production"}])
# resolved.tool_id, resolved.executor_type ('client' | 'http'), resolved.function

Returns list[ResolvedTool] (tool_id, version_number, executor_type, function). Raises API_ERROR (404) when any ref does not resolve.

tools.execute(tool_id, args, *, ...)

Run a tool's server-side http executor on the platform. The platform writes the tool span itself — do not report one for the same call.

out = await hub.tools.execute(
    tool_id, {"city": "Lahore"},
    alias="production", trace_id=trace_id, parent_span_id=llm_span_ref,
)
print(out.result, out.latency_ms, out.tool_version_id)

Parameter	Type	Notes
tool_id	str	From resolve(). Required (positional).
args	dict	The model's parsed arguments. Required (positional).
alias	str	Which alias to run.
version_number	int	Pin an exact version.
trace_id	str	Attach the span to this trace.
parent_span_id	str	Nest under this span (normally the llm span).

Returns ToolExecuteResult(result, status, latency_ms, tool_version_id).

tools.list(*, search=None, page=None, limit=None)

List tools for the team, newest first.

page = await hub.tools.list(search="weather", limit=10)

Parameter	Type	Notes
search	str	Free-text search on name.
page	int	1-based page.
limit	int	Page size.

Returns ToolListResult(data, total, page, limit).

tools.get(tool_id)

Fetch one tool's shell by id.

tool = await hub.tools.get(tool_id)

Returns ToolDetail (id, name, description, latest_version, created_at, updated_at).

tools.create(name, *, description=None)

Create a new tool shell. Commit a version with commit_version to give it a schema/executor.

tool = await hub.tools.create("get_weather", description="Weather lookup")

Returns ToolDetail.

tools.update(tool_id, **kwargs)

Update a tool's name and/or description. Does not touch versions.

updated = await hub.tools.update(tool_id, description="Updated description")

Returns ToolDetail.

tools.delete(tool_id)

Soft-delete a tool. Versions and aliases are preserved but unreachable. Returns None.

await hub.tools.delete(tool_id)

tools.commit_version(tool_id, *, description=None, parameters_schema, executor, changelog=None)

Commit a new immutable version for a tool. The first commit auto-creates production and staging aliases.

version = await hub.tools.commit_version(
    tool_id,
    description="Get weather for a city",
    parameters_schema={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
    executor={"type": "client"},
)

Returns ToolVersionDetail.

tools.list_versions(tool_id, *, page=None, limit=None)

List a tool's versions, newest first. Items omit parameters_schema/executor.

versions = await hub.tools.list_versions(tool_id)

Returns ToolVersionListResult(data, total, page, limit).

tools.get_version(tool_id, version_number)

Fetch one version with its full parameters_schema/executor.

v = await hub.tools.get_version(tool_id, 1)

Returns ToolVersionDetail.

tools.promote_alias(tool_id, alias, version_number)

Promote an alias to point at a specific version. Creates the alias if it doesn't exist.

alias = await hub.tools.promote_alias(tool_id, "production", 2)

Returns ToolAliasDetail (alias, version_number, tool_id).

tools.analytics(*, since=None, until=None)

Read aggregated call analytics (count, error rate, p50/p95 latency) per tool.

analytics = await hub.tools.analytics(since="2026-01-01T00:00:00Z")

Returns ToolAnalyticsResult (data).

@acrux.tool

Mark a function as a tool, deriving its name, description, and parameter schema from the function itself. Works bare (@acrux.tool) and called (@acrux.tool(alias="staging")). The function stays directly callable.

from acruxcore import acrux

@acrux.tool
async def get_weather(city: str) -> dict:
    """Get the current weather for a city.

    Args:
        city: City name, e.g. 'Lahore'.
    """
    return {"tempC": 21}

The decorator is pure — it performs no network calls and has no import-time side effects. The schema is derived from the signature; supported hints are str, int, float, bool, list[T], dict, Optional[T], Literal[...], and Enum subclasses. A parameter without a default is required, and each parameter's description comes from the docstring's Google-style Args: block.

Parameter	Type	Required	Notes
fn	Callable	(bare use)	The function, when used bare. Never pass explicitly.
name	str	no	Defaults to the function's name.
description	str	no	Defaults to the docstring's first paragraph.
parameters	dict	no	Hand-written JSON Schema; skips derivation. Use it for a type this converter cannot model.
alias	str	no	Catalog alias a sync moves. Default "production".
changelog	str	no	Release note for humans; never shown to the model.

A function with no docstring sends no description, which hands ownership of the model-facing text to the dashboard. Raises ToolSchemaError at decoration time when the schema cannot be derived (a missing hint, an unsupported type, or *args/**kwargs).

Structured output

response_format asks the model for a typed answer. Pass an OpenAI-shaped dict, or build one from a pydantic v2 model with pydantic_response_format — field-level Field(description=...) guidance reaches the model without hand-writing JSON Schema.

from pydantic import BaseModel, Field
from acruxcore import pydantic_response_format

class Sentiment(BaseModel):
    """Classify the message."""
    sentiment: str = Field(description="'pos', 'neg', or 'neutral'")

await hub.gateway.chat(
    "agent-model", messages,
    response_format=pydantic_response_format(Sentiment, name="sentiment"),
)

Pydantic is an optional dependency — pydantic_response_format() does not import it; the SDK resolves the schema at send time. The available shapes:

Variant	Shape
text	{"type": "text"}
json_object	{"type": "json_object"}
json_schema	{"type": "json_schema", "json_schema": {"name": ..., "schema": ..., "strict": ...}}
pydantic	pydantic_response_format(Model, *, name, strict=True)

The gateway forwards the format to each provider's native structured-output mode and relies on the provider to honour it — it does not validate the returned content against the schema, so parse and validate on your side when conformance matters. Raises PYDANTIC_NOT_AVAILABLE if a pydantic-built format is sent but pydantic is not installed.

BYO provider

Route a call directly to your own OpenAI-compatible endpoint instead of the gateway — the hop and its latency are skipped, and api_key is sent only to base_url, never to AcruxCore.

hub = AcruxCore(
    provider={"base_url": "https://api.groq.com/openai/v1", "api_key": os.environ["GROQ_API_KEY"]},
)

Pass provider to the constructor (a default for every call) or per-call to gateway.chat()/gateway.run_tool_loop(). There is no server-side catalog on this path, so every tool is sent inline as a full schema, and the SDK reports one llm span per round-trip. gateway.cost_usd and gateway.cache are always None (the gateway never saw the call). Raises PROVIDER_ERROR for a non-2xx provider response.

Span shapes

IngestSpan (the dict shape passed in spans to traces.ingest()):

Field	Type	Notes
spanId	str	Required; opaque, unique within the trace.
parentSpanId	str	Links to another span's spanId.
name	str	Required.
kind	str	'llm' | 'tool' | 'retrieval' | 'embedding' | 'agent' | 'chain' | 'other'.
status	str	'ok' | 'error' | 'unset'.
startTime	str	Required; ISO-8601 with offset.
endTime	str	ISO-8601.
model, provider	str	Model/provider metadata.
usage	dict	{promptTokens?, completionTokens?, totalTokens?, cachedTokens?}. cachedTokens is the part of promptTokens served from the provider's cache, priced at the cached rate.
costUsd	float	Cost in USD.
promptVersionId	str	Prompt lineage.
variables	Any	The prompt variables behind input. Read back when feedback becomes a dataset example.
input / output	any	Stored only with payload capture on.
attributes	dict	Free-form.
error	str	Error message for failed spans.

Error codes

All failures raise AcruxCoreError with a code attribute — use it for programmatic handling rather than matching on the message.

Code	When it is raised
MISSING_API_KEY	No api_key in args or env; or a BYO provider["api_key"] is empty.
MISSING_BASE_URL	No base_url in args or env; or a BYO provider["base_url"] is empty.
NETWORK_ERROR	All retries exhausted at the network level.
API_ERROR	Non-retryable HTTP error from the gateway (4xx, or 5xx after retries). Inspect status_code/body.
MISSING_VARIABLES	Template requires variables you did not supply.
TOOL_SCHEMA_ERROR	@acrux.tool: a missing hint, an unsupported type, or *args/**kwargs.
MISSING_DISPATCH	run_tool_loop: a tool has no implementation (raised before the first model call).
PYDANTIC_NOT_AVAILABLE	A pydantic-built response_format was sent but pydantic is not installed.
PROVIDER_ERROR	BYO: non-2xx response from your provider endpoint.

Sessions (hub.sessions)

sessions.list(*, page=None, limit=None)

List sessions with pagination.

page = await hub.sessions.list(limit=10)

Parameter	Type	Notes
page	int	1-based page.
limit	int	Page size.

Returns SessionListResult(data, total, page, limit).

sessions.get(session_id)

Get session detail with all traces.

session = await hub.sessions.get(session_id)

Returns SessionDetailResult (session, traces).

Evaluations

Datasets (hub.datasets)

Method	Description
datasets.create(name, *, overall_feedback=None)	Create a dataset. Returns DatasetDto.
datasets.build_from_feedback(name, feedback_ids, *, overall_feedback=None)	Build a dataset from trace feedback. Returns BuildFromFeedbackResult.
datasets.list()	List all datasets. Returns list[DatasetDto].
datasets.get(dataset_id)	Get dataset with examples. Returns DatasetWithExamples.
datasets.update(dataset_id, **kwargs)	Update dataset metadata (name, overall_feedback). Returns DatasetDto.
datasets.delete(dataset_id)	Delete a dataset. Returns None.
datasets.add_example(dataset_id, input, *, criteria=None, history=None)	Add an example. Returns DatasetExampleDto.
datasets.remove_example(dataset_id, example_id)	Remove an example. Returns None.

Experiments (hub.experiments)

Method	Description
experiments.create(dataset_id, version_ids, models, *, judge_model, prompt_id=None, name=None, alias=None, judge_prompt_id=None)	Create an experiment. judge_model is required: the model that grades every result. judge_prompt_id is an optional prompt that grades instead of the built-in judge. Returns ExperimentDto.
experiments.list()	List all experiments. Returns list[ExperimentDto].
experiments.get(experiment_id)	Get an experiment. Returns ExperimentDto.
experiments.start_run(experiment_id)	Start a run. Returns StartRunResult (run_id, status).

Runs (hub.runs)

Method	Description
runs.list(*, status=None, dataset_id=None, prompt_id=None, page=None, limit=None)	List runs. Returns RunListResponse.
runs.get(run_id)	Get run detail. Returns RunDetailDto.
runs.get_report(run_id)	Get run report. Returns RunReport.
runs.get_cell(run_id, cell_key)	Get a specific cell. Returns RunCellDetailDto.
runs.get_candidate(run_id, candidate_id)	Get a candidate. Returns CandidateDetail.
runs.promote_candidate(run_id, prompt_candidate_id, *, alias=None)	Promote a candidate. Returns PromoteResult.

Optimize (hub.optimize)

Method	Description
optimize.start(prompt_id, dataset_id, models, *, judge_model, draft_count=None, alias=None)	Start prompt optimization. judge_model is required. Returns StartOptimizeResult (run_id, status, prompt_mismatch_warning).

Where to next

• Quickstart — make your first call in ten minutes.
• Create a tool — put a tool in the catalog, from code or from the dashboard.
• Chat, stream, and collect feedback with the SDK.
• Node SDK reference — same methods, TypeScript style.