acruxcore 0.4.1

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.

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

render_prompt() returns RenderResult(messages, tools) — tools are any Tool Catalog tools attached to that prompt version, already in OpenAI shape. Feed them into run_tool_loop() to drive the whole tool-calling round-trip:

async def dispatch(name: str, args: dict):
    if name == "get_weather":
        return {"tempC": 18, "condition": "cloudy"}
    raise ValueError(f"Unknown tool: {name}")

result = await hub.run_tool_loop(
    model="gpt-4o",
    messages=render.messages,
    dispatch=dispatch,
    tools=render.tools,
)
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 dispatch

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 are dispatched concurrently (asyncio.gather); results are appended in call order, so dispatch must be safe to run in parallel. A dispatch 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 dispatch call, threaded into the same trace via the x-trace-id header. Turn it off with trace=False, or attach to an existing trace with trace={"trace_id": "..."}.

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, dispatch, *, ...)
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(*, ...)