acruxcore 0.5.0

Async Python SDK for Acrux Core — runtime prompt render, gateway chat, tool loops, traces, and feedback

acruxcore (Python)

Async Python SDK for Acrux Core. Fetch rendered prompts at runtime, call the AI gateway, run client-side tool loops, and report/read traces — full feature parity with the TypeScript SDK, with a Pythonic async/await API.

Installation

pip install acruxcore

Requires Python 3.9+. Depends only on httpx.

Using Node or TypeScript instead? Install the JavaScript SDK from npm: npm install @acruxcoreai/sdk — see @acruxcoreai/sdk on npm.

Changelog

0.5.0 — breaking

• tools= changed meaning. It now takes functions decorated with @acrux.tool. Raw OpenAI-shaped dicts move to tool_defs=: run_tool_loop(model, messages, dispatch=d, tools=raw) → run_tool_loop(model, messages, dispatch=d, tool_defs=raw).
• dispatch is now keyword-only and optional. It was the third positional argument. A decorated tool runs its own function; a tool_refs entry with an http executor runs on the platform. You still need dispatch for tool_defs, and for a client ref you have not decorated — that case now raises MISSING_DISPATCH before the first model call rather than mid-loop.
• run_tool_loop makes catalog requests before the first completion when tools= is given: one sync per tool, cached per process. Pass sync=False if a deploy step already synced them.
• New: @acrux.tool — name, description and parameter schema derived from the function itself. See Tools.
• New: hub.tools.sync(), hub.tools.resolve(), hub.tools.execute().

Quickstart

import asyncio
from acruxcore import AcruxCore

async def main():
    async with AcruxCore(
        api_key="...",                               # or env ACRUXCORE_API_KEY
        base_url="https://api.acruxcore.com/api/v1",  # or env ACRUXCORE_BASE_URL
    ) as hub:
        result = await hub.render_prompt("summarise-article", "production", {"article": "..."})
        print(result.messages)

asyncio.run(main())

AcruxCore owns an httpx.AsyncClient, so use it as an async context manager (async with) or call await hub.aclose() when done. Create one instance at startup and reuse it — the render cache is a process-wide singleton.

Chat

chat() is a single, non-looping call to the gateway's OpenAI-compatible POST /gateway/chat/completions. It routes to the right provider, prices the call, and records a trace server-side.

r = await hub.chat("gpt-4o-mini", [{"role": "user", "content": "Say hi in one word."}])
print(r.content)        # 'Hello!'
print(r.finish_reason)  # 'stop'
print(r.usage)          # ChatUsage(prompt_tokens=..., completion_tokens=..., total_tokens=...)
print(r.gateway)        # GatewayCallMeta(request_id=..., provider=..., cost_usd=..., cache=...)

Pass tools= / tool_refs= / tool_choice= just like the raw endpoint. If the model calls a tool, chat() hands it back raw on r.message["tool_calls"] — it never dispatches. Use run_tool_loop() for that.

Streaming

Pass stream=True to get an async iterator of chunks:

async for chunk in await hub.chat("gpt-4o-mini", messages, stream=True):
    print(chunk.delta.get("content", ""), end="", flush=True)
    if chunk.finish_reason:
        print(f"\n(done: {chunk.finish_reason})")

Each chunk mirrors one chat.completion.chunk SSE frame (id, model, delta, finish_reason); iteration ends when the gateway sends data: [DONE].

Tools

Decorate a function with @acrux.tool and hand it to the loop. The name, the model-facing description and the parameter schema all come from the function, so there is nothing to keep in sync by hand:

import httpx
from acruxcore import AcruxCore, acrux

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

    Args:
        city: City name, e.g. 'Lahore'.
    """
    async with httpx.AsyncClient() as http:
        res = await http.get(f"https://wttr.in/{city}", params={"format": "j1"})
    current = res.json()["current_condition"][0]
    return {"city": city, "temp_c": int(current["temp_C"])}


async with AcruxCore() as hub:
    result = await hub.run_tool_loop(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Should I run in Lahore this evening?"}],
        tools=[get_weather],
    )
    print(result.content)      # final assistant text
    print(result.messages)     # full transcript, incl. tool calls/results
    print(result.iterations)   # number of model round-trips
    print(result.trace_id)     # trace covering every round-trip + tool call

The decorator is pure: it attaches a spec to the function and returns it unchanged, so await get_weather(city="London") still works in a test.

What the decorator derives

These rules are the SDK's contract, so they are worth knowing exactly:

• Name — the function name.
• Description — the docstring's first paragraph. A function with no docstring sends no description, which leaves whatever your team wrote in the dashboard in place. Write one and code owns it: every sync overwrites the dashboard's text. Pick per tool which side owns the wording.
• Parameter descriptions — the Args: block, Google style.
• Required — every parameter without a default.
• Supported hints — str, int, float, bool, list[T], dict, Optional[T], Literal[...], and Enum subclasses. Anything else raises ToolSchemaError at decoration time — at import, not mid-run.

@acrux.tool(parameters={...}) is the escape hatch: pass a JSON Schema and the derivation is skipped entirely.

@acrux.tool(parameters={"type": "object", "properties": {"table": {"type": "string"}}, "required": ["table"]})
async def count_rows(table: str) -> dict:
    ...

On Python 3.9 a tool signature must spell an optional parameter Optional[int] rather than int | None; the X | Y form in an annotation is 3.10+.

The catalog round-trip

On the first call, run_tool_loop syncs each decorated tool into the Tool Catalog and then passes it to the model as a tool_refs entry rather than as an inline schema. So the schema the model sees is the one the catalog holds, the dashboard shows a version history for a tool defined in code, and every tool span records the exact version that ran. The sync is idempotent and cached per process: an unchanged tool costs one request per process, a changed one commits a new version and moves its alias. Pass sync=False when a deploy step already synced them.

Catalog tools you didn't decorate

A tool whose catalog version has an http executor needs no local code at all. Name it in tool_refs= and the platform calls the endpoint, writes the tool span with the real payloads, and hands the result back to the loop:

result = await hub.run_tool_loop(
    model="gpt-4o-mini",
    messages=messages,
    tool_refs=[{"name": "search_orders", "alias": "production"}],
)

dispatch is still there, and is what you need for two cases: raw OpenAI-shaped dicts passed as tool_defs=, and a tool_refs entry with a client executor you have not decorated. Something has to run a client tool, so if neither a decorated function nor dispatch can, the loop raises MISSING_DISPATCH before the first model call — the failure costs no tokens.

async def dispatch(name: str, args: dict):
    if name == "get_weather":
        return await fetch_weather_from_your_provider(args["city"])
    raise ValueError(f"Unknown tool: {name}")

result = await hub.run_tool_loop(
    model="gpt-4o-mini", messages=messages, tool_defs=raw_defs, dispatch=dispatch
)

Prompt-attached tools arrive this way too: render_prompt() returns RenderResult(messages, tools) where tools are the version's attached catalog tools in OpenAI shape — those go in tool_defs=.

The loop's behaviour

run_tool_loop() stops when the model responds without calling a tool, or after max_iterations round-trips (default 10; result.stopped_at_limit is True then). When the model requests several tools in one turn they run concurrently (asyncio.gather); results are appended in call order, so a tool body must be safe to run in parallel. A tool that raises is not caught — wrap it yourself if you want a tool failure reported back to the model as a tool-result message instead of aborting the loop.

The loop auto-reports one trace: the gateway records an llm span per round-trip, and the SDK adds a tool span per client-side call, threaded into the same trace via the x-trace-id header. Tools that ran on the platform get their span from the platform, so they land in the same waterfall without being reported twice. Turn tracing off with trace=False, or attach to an existing trace with trace={"trace_id": "..."}.

Catalog access without the loop

hub.tools reaches the catalog directly — useful in a deploy step, or when you drive the model yourself:

await hub.tools.sync([get_weather], on_conflict="error")   # reconcile at deploy time
resolved = await hub.tools.resolve([{"name": "search_orders"}])
out = await hub.tools.execute(resolved[0].tool_id, {"query": "refunds"})

sync returns, per tool, the version it landed on and whether this call committed it. on_conflict="error" raises when a commit supersedes a version someone edited in the dashboard; the default warns instead, so a dashboard experiment can never block a deploy.

Reporting traces

from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()

res = await hub.trace({
    "name": "support-agent-run",
    "spans": [
        {"spanId": "s1", "name": "gpt-4o-mini", "kind": "llm", "startTime": now, "endTime": now,
         "model": "gpt-4o-mini", "usage": {"promptTokens": 120, "completionTokens": 40, "totalTokens": 160}},
        {"spanId": "s2", "parentSpanId": "s1", "name": "search_docs", "kind": "tool",
         "startTime": now, "attributes": {"query": "refunds"}},
    ],
})

# Append another span to the same trace later:
await hub.trace({"traceId": res.trace_id, "spans": [
    {"spanId": "s3", "parentSpanId": "s1", "name": "finalize", "kind": "chain", "startTime": now}]})

kind is one of llm | tool | retrieval | embedding | agent | chain | other; status is ok | error | unset. input/output are stored only when your team has payload capture on (or you pass capturePayloads: True). Up to 200 spans per call. Span keys are camelCase (spanId, parentSpanId, startTime) because they are sent to the API verbatim.

Feedback

fb = await hub.submit_feedback(
    trace_id,
    rating=-1,                # -1..5
    label="wrong_answer",
    comment="The tool call missed relevant docs.",
    source="end_user",        # 'user' | 'developer' | 'end_user' | 'api'
)

await hub.submit_feedback(trace_id, span_id="s1", rating=5)  # scope to one span

# Edit later (author only). Pass a value to change, None to clear, omit to keep:
await hub.update_feedback(trace_id, fb.id, rating=1)

At least one of rating / label / comment is required per call.

Reading traces back

detail = await hub.get_trace(trace_id)
print(detail.trace.status, detail.trace.total_cost_usd, detail.trace.total_tokens)
print(detail.spans[0].model, detail.spans[0].latency_ms)

page = await hub.list_traces(session_id="tokyo-trip-plan-01", limit=10)

Configuration

Argument	Environment Variable	Default	Description
api_key	ACRUXCORE_API_KEY	required	Your Acrux Core API key
base_url	ACRUXCORE_BASE_URL	required	API base URL (e.g. https://api.acruxcore.com/api/v1)
cache_ttl	—	60000 (60s)	Milliseconds before a cached render is stale
max_cache_size	—	500	Max prompt entries in the in-process LRU cache
max_retries	—	1	Retries on transient failure (2 total attempts)
retry_interval	—	500	Milliseconds between retries
timeout	—	30	Per-request timeout, in seconds

Error handling

from acruxcore import AcruxCoreError

try:
    await hub.render_prompt("my-prompt", "production", vars)
except AcruxCoreError as err:
    if err.code == "MISSING_VARIABLES":
        print("Missing template variables:", err.body["error"]["missing"])
    elif err.code == "NETWORK_ERROR":
        print("Acrux Core API unreachable. Check base_url.")
    elif err.code == "API_ERROR":
        print(f"Acrux Core API error {err.status_code}")
    raise

Error codes: MISSING_API_KEY, MISSING_BASE_URL, NETWORK_ERROR, API_ERROR, MISSING_VARIABLES.

Caching

• Cache key: {api_key}:{prompt_name}:{alias} — scoped per team, prompt, alias.
• Variables are not part of the key — different variable values share a slot.
• Stale-while-revalidate: a stale hit returns the cached value immediately and fires a background refresh (asyncio task).
• API unreachable + stale entry: serves stale and logs a warning.
• API unreachable + cold cache: raises AcruxCoreError(code="NETWORK_ERROR").

Method parity with the TypeScript SDK

TypeScript	Python
renderPrompt(name, alias, vars)	render_prompt(name, alias, variables)
chat({...})	chat(model, messages, *, ...)
chat({stream: true})	chat(..., stream=True) → async iterator
runToolLoop({...})	run_tool_loop(model, messages, *, tools=, tool_defs=, tool_refs=, dispatch=None, sync=True, ...)
acrux.tool({name, parameters}, handler)	@acrux.tool (or @acrux.tool(parameters={...}))
hub.tools.sync(tools, {onConflict})	hub.tools.sync(tools, on_conflict=...)
hub.tools.resolve(refs)	hub.tools.resolve(refs)
hub.tools.execute(toolId, args, {...})	hub.tools.execute(tool_id, args, ...)
trace(input)	trace(input)
submitFeedback({...})	submit_feedback(trace_id, *, ...)
updateFeedback({...})	update_feedback(trace_id, feedback_id, *, ...)
getTrace(id)	get_trace(trace_id)
listTraces({...})	list_traces(*, ...)