coalex

Python SDK for Coalex — AI governance observability for enterprises in regulated markets.

Built on OpenTelemetry.

Install

pip install coalex
# or
uv add coalex

Quick start

import coalex

coalex.register(api_key=os.environ["COALEX_API_KEY"], service_name="my-agent")
coalex.auto_instrument()

# `version` is the build this agent is running. It is optional today and omitting it warns:
# a trace with no version cannot be attributed to a build, and the dashboard
# records it as `unknown`. It becomes required in coalex 2.0.
with coalex.coalex_context(agent_id="support-agent", version="1.4.0"):
    answer = my_agent.run(question)

decision = coalex.evaluate(
    request_id=request_id,
    input={"question": question},
    output={"answer": answer},
    # The metric catalogue is validated before the call: f1, word_overlap, bleu, rouge_l,
    # exact_match, levenshtein, contains. This line used to read ["faithfulness"], which is not
    # one of them — so the first example a customer copied raised ValueError.
    metrics={"answer": ["f1", "word_overlap"]},
)

Several agents in one process

A context opened inside another agent's context belongs to the inner agent: its spans carry coalex.parent_agent_id (the enclosing agent) and the enclosing request_id / run_id, but not the enclosing agent's version or prompt. coalex.agent_node wraps a LangGraph node function (sync or async) so each node is its own agent, re-opened on every resume, and a LangGraph pause (interrupt()) ends the span OK rather than as an error. current_agent_id(), current_parent_agent_id(), current_request_id(), current_run_id() and current_agent_version() read the active context.

graph.add_node("rfq", coalex.agent_node("rfq-worker", rfq_worker, version="1.4.0"))

with coalex.coalex_context(agent_id="orchestrator", request_id=request_id, run_id=run_id, version="2.0.0"):
    graph.compile(checkpointer=checkpointer).invoke(state, config)

Holding an action until Coalex decides

coalex.govern (extra langchain) holds a LangGraph agent's write until Coalex releases it. Mark each tool metadata={"coalex": {"kind": "read" | "write"}}; a tool without a marker is held as a write. Run the agent under a checkpointer with a thread_id, inside its coalex_context, and give it the resume hook Coalex calls once a person decides (resume_hook=, a ResumeHook or a function returning one); a write without a hook URL does not run.

from coalex.govern import GovernMiddleware, govern

agent = create_agent(model, tools, middleware=[GovernMiddleware(resume_hook=hook)], checkpointer=checkpointer)


def pay(state):  # a flow node: at most one governed write per node
    return {"receipt": govern(send_wire, state["wire"], resume_hook=hook)}

An escalation pauses the run (interrupt({"escalation_id", "tool", "args"})). Resume it with the body Coalex posts to the agent, Command(resume=body): approved and corrected run the tool, rejected and expired do not. When Coalex cannot decide, the tool does not run and GovernError says why. Keep the model to one tool call per turn (parallel_tool_calls=False): one resume value answers one held call.

Replaying historical cases

A replay re-runs an agent over past cases and scores it against the decisions people took. Each case runs inside a context that names the replay, the case and when the case happened; every span inside, nested agents and auto-instrumented model and tool calls included, carries coalex.replay_id, coalex.case_id and coalex.case_timestamp (ISO 8601 with offset), and none of them is set outside a replay.

with coalex.coalex_context(
    agent_id="triage",
    version="1.4.0",
    run_id=replay_id,
    replay_id=replay_id,
    case_id=case.id,
    case_timestamp=case.timestamp,  # timezone-aware datetime
):
    graph.invoke(case.input)

Diagnostics

The SDK logs under the coalex logger and never configures logging for you — a library that calls basicConfig takes over your application's log configuration, which is not its to take. It attaches a NullHandler and leaves the rest to you.

To see what it is doing:

import logging

logging.basicConfig(level=logging.INFO)  # your application's own setup
logging.getLogger("coalex").setLevel(logging.DEBUG)

Without raising the level you still get warnings and errors — but only through Python's lastResort handler if your application configures none of its own, which prints with no timestamp, level or logger name and cannot be filtered.

No API key, proxy token or upstream credential reaches a log line or an exception message. Raising the level does not change that: it is a property of what is written, not of what is filtered.

Telemetry never breaks your agent

A failing Coalex endpoint drops spans; it does not raise into your code. evaluate() and resolve() do raise, because a governance decision you asked for and did not get is not something to swallow.

TypeScript

There is a TypeScript SDK, and it is smaller. See Python vs TypeScript before choosing.

Documentation

https://docs.coalex.ai