Skip to content

SDK Reference

Complete API reference for the Coalex SDKs:

  • Python SDK (coalex v1.9.0) -- published on PyPI
  • TypeScript SDK (@coalex-ai/sdk v1.7.0) -- published on npm

Installation

pip install coalex                    # core SDK
pip install coalex[auto-instrument]   # + all auto-instrumentation
pip install coalex[openai]            # + OpenAI only

npm install @coalex-ai/sdk            # core SDK
npm install @arizeai/openinference-instrumentation-openai  # + OpenAI auto-instrumentation

See Installation for all extras and framework-specific options.

Usage Pattern

Every Coalex integration follows the same pattern:

graph LR
    A[register] --> B[auto_instrument]
    B --> B2[declare_agent]
    B2 --> C[coalex_context]
    C --> D[evaluate]
    D --> E[resolve]

1. Register -- connect to Coalex

Call once at application startup. Configures the global TracerProvider with an OTLP exporter pointing at your Coalex proxy.

import coalex

coalex.register(
    endpoint="https://traces.coalex.ai",
    api_key="your-api-key",
    service_name="my-agent",
)

import { register } from "@coalex-ai/sdk";

register({
    endpoint: "https://traces.coalex.ai",
    apiKey: "your-api-key",
    serviceName: "my-agent",
});

2. Auto-instrument -- patch LLM frameworks

Automatically patches all installed LLM libraries (OpenAI, LangChain, LlamaIndex, etc.) to emit OpenInference spans.

results = coalex.auto_instrument()
# {"openai": "success", "langchain": "not_installed", ...}

import { autoInstrument } from "@coalex-ai/sdk";

const results = autoInstrument();
// { openai: "success", langchain: "not_installed", ... }

3. Wrap with context -- tag every span

All spans created inside a coalex_context block inherit the agent metadata, making them queryable in the dashboard.

with coalex.coalex_context(agent_id="support-bot", request_id="req-123"):
    response = client.chat.completions.create(...)

import { coalexContext } from "@coalex-ai/sdk";

await coalexContext({ agentId: "support-bot", requestId: "req-123" }, async () => {
    const response = await client.chat.completions.create(...);
});

4. Declare agent -- register before traces

Pre-register your agent so the dashboard recognizes it before traces arrive.

agent = coalex.declare_agent(agent_id="support-bot", display_name="Support Bot")
# AgentDeclaration(agent_id="support-bot", lifecycle="active", created=True)

import { declareAgent } from "@coalex-ai/sdk";

const agent = await declareAgent({ agentId: "support-bot", displayName: "Support Bot" });
// { agentId: "support-bot", lifecycle: "active", created: true }

5. Evaluate -- assess risk

Submit agent output for automated risk assessment. Low-risk outputs are auto-approved; high-risk ones are escalated for human review.

decision = coalex.evaluate(
    request_id="req-123",
    input={"question": "What is the policy?"},
    output={"answer": "The policy states..."},
    metrics={"answer": ["f1", "rouge_l"]},
)

import { evaluate } from "@coalex-ai/sdk";

const decision = await evaluate({
    requestId: "req-123",
    input: { question: "What is the policy?" },
    output: { answer: "The policy states..." },
    metrics: { answer: ["f1", "rouge_l"] },
});

6. Resolve -- human-in-the-loop

When an output is escalated, a human reviewer approves, rejects, or corrects it.

result = coalex.resolve(
    escalation_id=decision.escalation_id,
    decision="approved",
    reviewer={"name": "Dr. Smith", "email": "dr.smith@hospital.org"},
    reason="Output is clinically accurate.",
)

import { resolve } from "@coalex-ai/sdk";

const result = await resolve({
    escalationId: decision.escalationId,
    decision: "approved",
    reviewer: { name: "Dr. Smith", email: "dr.smith@hospital.org" },
    reason: "Output is clinically accurate.",
});

Public API

Core Functions

Function Description Reference
register() Configure the OTLP exporter and global TracerProvider Details
declare_agent() Pre-register an agent before traces arrive Details
coalex_context() Create a parent span with agent metadata Details
auto_instrument() Patch all installed LLM frameworks Details
get_prompt() Fetch prompt templates from Prompt Vault (Python) Details
evaluate() Submit output for risk-based evaluation Details
resolve() Submit human review for an escalation Details

Data Classes

Class Description Reference
AgentDeclaration Result of declare_agent() Details
PromptVersion Result of get_prompt() Details
EvaluationDecision Result of evaluate() Details
ResolutionResult Result of resolve() Details
MetricResult Per-field metric scores Details

Extension Decorators / Wrappers

Instrument custom pipeline steps without LlamaIndex or LangChain. See Extensions.

Python Decorator TypeScript Wrapper Span Kind Reference
@retrieval_span retrievalSpan() RETRIEVER Details
@embedding_span embeddingSpan() EMBEDDING Details
@reranker_span rerankerSpan() RERANKER Details
@tool_span toolSpan() TOOL Details
@guardrail_span guardrailSpan() GUARDRAIL Details

Full Example

import os
import coalex
from coalex.ext.retrieval import retrieval_span, Document
from openai import OpenAI

# 1. Register
coalex.register(
    endpoint=os.environ["COALEX_ENDPOINT"],
    api_key=os.environ["COALEX_API_KEY"],
    service_name="medical-agent",
)

# 2. Auto-instrument
coalex.auto_instrument()

# 3. Declare agent (recommended)
coalex.declare_agent(agent_id="medical-bot", display_name="Medical Bot")


# 4. Custom retrieval step
@retrieval_span(name="knowledge_base", query_arg="query")
def retrieve(query: str) -> list[Document]:
    # Your retrieval logic here
    return [Document(content="...", id="doc-1", score=0.95)]


# 5. Agent invocation
client = OpenAI()

with coalex.coalex_context(agent_id="medical-bot", request_id="req-456"):
    docs = retrieve(query="What are the side effects?")
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Based on: {docs[0].content}"}],
    )
    answer = response.choices[0].message.content

    # 6. Evaluate
    decision = coalex.evaluate(
        request_id="req-456",
        input={"question": "What are the side effects?"},
        output={"answer": answer},
        metrics={"answer": ["f1", "rouge_l"]},
    )

    # 7. Resolve (if escalated)
    if decision.status == "escalated":
        result = coalex.resolve(
            escalation_id=decision.escalation_id,
            decision="approved",
            reviewer={"name": "Dr. Smith", "email": "dr.smith@hospital.org"},
        )

import { register, coalexContext, autoInstrument, declareAgent, evaluate, resolve } from "@coalex-ai/sdk";
import { retrievalSpan, type Document } from "@coalex-ai/sdk/ext";
import OpenAI from "openai";

// 1. Register
register({
    endpoint: process.env.COALEX_ENDPOINT!,
    apiKey: process.env.COALEX_API_KEY!,
    serviceName: "medical-agent",
});

// 2. Auto-instrument
autoInstrument();

// 3. Declare agent (recommended)
await declareAgent({ agentId: "medical-bot", displayName: "Medical Bot" });

// 4. Custom retrieval step
const retrieve = retrievalSpan(
    { name: "knowledge_base" },
    async (query: string): Promise<Document[]> => {
        // Your retrieval logic here
        return [{ content: "...", id: "doc-1", score: 0.95 }];
    },
);

// 5. Agent invocation
const client = new OpenAI();

await coalexContext({ agentId: "medical-bot", requestId: "req-456" }, async () => {
    const docs = await retrieve("What are the side effects?");
    const response = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: `Based on: ${docs[0].content}` }],
    });
    const answer = response.choices[0].message.content;

    // 6. Evaluate
    const decision = await evaluate({
        requestId: "req-456",
        input: { question: "What are the side effects?" },
        output: { answer },
        metrics: { answer: ["f1", "rouge_l"] },
    });

    // 7. Resolve (if escalated)
    if (decision.status === "escalated") {
        const result = await resolve({
            escalationId: decision.escalationId!,
            decision: "approved",
            reviewer: { name: "Dr. Smith", email: "dr.smith@hospital.org" },
        });
    }
});

API Reference

coalex

Coalex Python SDK — AI governance observability.

Classes

AgentDeclaration dataclass

Result of a declare_agent() call.

Source code in coalex/agents.py 
14
15
16
17
18
19
20
@dataclasses.dataclass(frozen=True)
class AgentDeclaration:
    """Result of a declare_agent() call."""

    agent_id: str
    lifecycle: str  # "declared" | "active"
    created: bool

ConsoleAuthError

Bases: Exception

A call did not carry a valid Console signature. Never says which check failed, to the caller.

The reason is available to the agent's own logs via reason; it must not travel back to whoever made the call, or the error becomes an oracle for guessing a valid token.

Source code in coalex/console_auth.py 
46
47
48
49
50
51
52
53
54
55
class ConsoleAuthError(Exception):
    """A call did not carry a valid Console signature. Never says which check failed, to the caller.

    The reason is available to the agent's own logs via ``reason``; it must not travel back to
    whoever made the call, or the error becomes an oracle for guessing a valid token.
    """

    def __init__(self, reason: str) -> None:
        super().__init__("unauthorized")
        self.reason = reason

ConsoleCaller dataclass

Who the Console says this turn belongs to, once the signature has been verified.

Source code in coalex/console_auth.py 
70
71
72
73
74
75
76
77
78
@dataclasses.dataclass(frozen=True)
class ConsoleCaller:
    """Who the Console says this turn belongs to, once the signature has been verified."""

    account_id: str
    agent_id: str
    user_id: str | None = None
    user_email: str | None = None
    claims: Mapping[str, Any] = dataclasses.field(default_factory=dict)

MissingExpectedAccountWarning

Bases: UserWarning

verify_console_call() was called without expected_account (it becomes required in 2.0).

A subclass of UserWarning so CPython's default filters show it, and a named class so silencing it is a typed one-liner::

warnings.filterwarnings("ignore", category=coalex.MissingExpectedAccountWarning)

Reported against the caller's frame, so a filter keyed on module names theirs, not ours.

Source code in coalex/console_auth.py 
58
59
60
61
62
63
64
65
66
67
class MissingExpectedAccountWarning(UserWarning):
    """``verify_console_call()`` was called without ``expected_account`` (it becomes required in 2.0).

    A subclass of ``UserWarning`` so CPython's default filters show it, and a **named** class so
    silencing it is a typed one-liner::

        warnings.filterwarnings("ignore", category=coalex.MissingExpectedAccountWarning)

    Reported against the caller's frame, so a filter keyed on ``module`` names theirs, not ours.
    """

ConsoleKeys

Bases: Mapping[str, str]

The Console's public keys by kid, fetched on demand and cached.

Reads are what verify_console_call does: kid in keys then keys[kid]. Both may fetch, so both may block for up to timeout_seconds. In an async agent, call :meth:refresh at startup — after that the steady state is a dict lookup, and a fetch only happens once per TTL or when a key id is seen that is not held.

A fetch that fails leaves the previous keys in place. Serving a slightly stale key is better than refusing every call because the Console was briefly unreachable; a key that is genuinely gone stops verifying anyway, because the signature will not match.

Source code in coalex/console_jwks.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class ConsoleKeys(Mapping[str, str]):
    """The Console's public keys by ``kid``, fetched on demand and cached.

    Reads are what ``verify_console_call`` does: ``kid in keys`` then ``keys[kid]``. Both may fetch,
    so both may block for up to ``timeout_seconds``. In an async agent, call :meth:`refresh` at
    startup — after that the steady state is a dict lookup, and a fetch only happens once per TTL or
    when a key id is seen that is not held.

    A fetch that fails leaves the previous keys in place. Serving a slightly stale key is better
    than refusing every call because the Console was briefly unreachable; a key that is genuinely
    gone stops verifying anyway, because the signature will not match.
    """

    def __init__(
        self,
        jwks_url: str,
        *,
        ttl_seconds: float = DEFAULT_TTL_SECONDS,
        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
        client: httpx.Client | None = None,
    ) -> None:
        self._url = jwks_url
        self._ttl = ttl_seconds
        self._timeout = timeout_seconds
        self._client = client
        self._keys: dict[str, str] = {}
        self._fetched_at: float | None = None
        # Separate from any other fetch: the floor below must throttle *repeated* unknown-kid
        # refetches, not be armed by the routine one that populated the cache a moment ago.
        self._last_unknown_refetch: float = float("-inf")
        # When the last fetch ATTEMPT was, and how long to wait before the next one.
        #
        # `_fetch_locked` deliberately leaves `_fetched_at` alone on failure, so the cache never
        # claims to hold something it does not. The TTL check then read as "never fetched" on every
        # single call, and each `kid in keys` made a fresh HTTP GET with a five-second timeout —
        # under `self._lock`, so concurrent verifications did not merely repeat the cost, they
        # queued behind it. In the demo agent that verification is synchronous inside the request
        # handler, so a Console outage became a serialised five-second wait per request.
        #
        # The unknown-kid throttle did not cover this branch: it is armed only when a kid is missing
        # from a cache that was successfully populated.
        self._last_attempt: float = float("-inf")
        self._backoff: float = 0.0
        # Two requests arriving together must not both fetch, and must not see a half-built dict.
        self._lock = threading.Lock()

    @property
    def url(self) -> str:
        return self._url

    def refresh(self) -> dict[str, str]:
        """Fetch now, regardless of the TTL. Returns the keys held afterwards.

        Never raises: a failure logs and keeps whatever was already cached. The caller cannot do
        anything useful with the exception — a missing key surfaces later as a refused token, which
        is the correct failure and the one the verifier already reports.
        """
        with self._lock:
            return self._fetch_locked()

    def _fetch_locked(self) -> dict[str, str]:
        self._last_attempt = time.monotonic()
        try:
            if self._client is not None:
                response = self._client.get(self._url, timeout=self._timeout)
            else:
                response = httpx.get(self._url, timeout=self._timeout)
            response.raise_for_status()
            document = response.json()
        except Exception as exc:  # noqa: BLE001 - transport, status and JSON failures are one case
            logger.warning("could not fetch Console keys from %s: %s", self._url, exc)
            self._widen_backoff()
            return dict(self._keys)

        if isinstance(document, (str, bytes)):  # a server that sent JSON as a string
            try:
                document = json.loads(document)
            except ValueError:
                document = None

        keys = document.get("keys") if isinstance(document, dict) else None
        if not isinstance(keys, list):
            logger.warning("Console key set at %s has no 'keys' array", self._url)
            self._widen_backoff()
            return dict(self._keys)

        fetched: dict[str, str] = {}
        for entry in keys:
            if not isinstance(entry, Mapping):
                continue
            kid = entry.get("kid")
            pem = _jwk_to_pem(entry)
            if isinstance(kid, str) and pem:
                fetched[kid] = pem

        if not fetched:
            # An empty set is what the Console publishes when it has no signing key — a real state,
            # but not a reason to discard keys that are working here.
            logger.warning("Console key set at %s contained no usable Ed25519 keys", self._url)
            self._widen_backoff()
            return dict(self._keys)

        self._keys = fetched
        self._fetched_at = time.monotonic()
        self._backoff = 0.0
        return dict(self._keys)

    def _widen_backoff(self) -> None:
        """Double the wait after a failed attempt, from the unknown-kid floor up to one TTL.

        Capped at the TTL because that is already the rhythm this cache keeps when everything works:
        waiting longer than a routine refresh to retry would mean a Console that came back stayed
        unnoticed for longer than a key rotation would.
        """
        ceiling = max(self._ttl, MIN_REFETCH_INTERVAL_SECONDS)
        self._backoff = min(max(self._backoff * 2, MIN_REFETCH_INTERVAL_SECONDS), ceiling)

    def _ensure(self, kid: str | None = None) -> None:
        with self._lock:
            now = time.monotonic()
            if self._fetched_at is None or (now - self._fetched_at) >= self._ttl:
                # Due for a fetch — but not if the last attempt failed and the backoff has not
                # elapsed. Whatever is cached keeps being served meanwhile, which is the
                # property this class already promised: a slightly stale key beats refusing every
                # call because the Console was briefly unreachable.
                if (now - self._last_attempt) < self._backoff:
                    return
                self._fetch_locked()
                return

            if kid is None or kid in self._keys:
                return

            # A key id we do not hold: either a rotation to catch up with, or a token that will be
            # refused whatever we do. Fetch for the first one, throttle the second — the floor is
            # measured from the last *unknown-kid* refetch, so a genuinely new kid is picked up on
            # its first appearance rather than after the routine fetch happens to age out.
            if (now - self._last_unknown_refetch) < MIN_REFETCH_INTERVAL_SECONDS:
                return
            self._last_unknown_refetch = now
            self._fetch_locked()

    # --- Mapping ----------------------------------------------------------

    def __getitem__(self, kid: str) -> str:
        self._ensure(kid)
        with self._lock:
            return self._keys[kid]

    def __contains__(self, kid: object) -> bool:
        if not isinstance(kid, str):
            return False
        self._ensure(kid)
        with self._lock:
            return kid in self._keys

    def __iter__(self) -> Iterator[str]:
        self._ensure()
        with self._lock:
            return iter(dict(self._keys))

    def __len__(self) -> int:
        self._ensure()
        with self._lock:
            return len(self._keys)
Methods:
refresh 
refresh() -> dict[str, str]

Fetch now, regardless of the TTL. Returns the keys held afterwards.

Never raises: a failure logs and keeps whatever was already cached. The caller cannot do anything useful with the exception — a missing key surfaces later as a refused token, which is the correct failure and the one the verifier already reports.

Source code in coalex/console_jwks.py 
122
123
124
125
126
127
128
129
130
def refresh(self) -> dict[str, str]:
    """Fetch now, regardless of the TTL. Returns the keys held afterwards.

    Never raises: a failure logs and keeps whatever was already cached. The caller cannot do
    anything useful with the exception — a missing key surfaces later as a refused token, which
    is the correct failure and the one the verifier already reports.
    """
    with self._lock:
        return self._fetch_locked()

MissingAgentVersionWarning

Bases: UserWarning

Raised by coalex_context() when no version was given.

A subclass of UserWarning so CPython's default filters show it, and a named class so silencing it is a typed one-liner rather than a regex over our prose::

warnings.filterwarnings("ignore", category=coalex.MissingAgentVersionWarning)

module="coalex.*" does not work, and that is a consequence of getting the blame right: the warning is reported against the caller's frame, so a filter keyed on the module names theirs.

Source code in coalex/context.py 
68
69
70
71
72
73
74
75
76
77
78
class MissingAgentVersionWarning(UserWarning):
    """Raised by ``coalex_context()`` when no ``version`` was given.

    A subclass of ``UserWarning`` so CPython's default filters show it, and a **named** class so
    silencing it is a typed one-liner rather than a regex over our prose::

        warnings.filterwarnings("ignore", category=coalex.MissingAgentVersionWarning)

    ``module="coalex.*"`` does not work, and that is a consequence of getting the blame right: the
    warning is reported against the caller's frame, so a filter keyed on the module names *theirs*.
    """

ComposedQuestion dataclass

What a Question Composer decides the card shows (spec §6).

Everything the approver reads, and nothing else: the question, the curated summary, the values, and the schema that says how each value is answered. A composer that returns these four has decided the whole card.

Source code in coalex/decision_request.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class ComposedQuestion:
    """What a Question Composer decides the card shows (spec §6).

    Everything the approver reads, and nothing else: the question, the curated summary, the values,
    and the schema that says how each value is answered. A composer that returns these four has
    decided the whole card.
    """

    question: str
    summary: str | None
    output: dict[str, Any]
    schema: QuestionSchema

ContextPayload dataclass

What the human needs to judge (spec §4.1, §6 de-biasing).

Deliberately excludes the agent's reasoning, confidence, and risk score — those stay in the trace for compliance and never reach the approval card. output is shown as an editable answer, not as persuasion.

Source code in coalex/decision_request.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@dataclasses.dataclass(frozen=True)
class ContextPayload:
    """What the human needs to judge (spec §4.1, §6 de-biasing).

    Deliberately excludes the agent's reasoning, confidence, and **risk score** — those
    stay in the trace for compliance and never reach the approval card. ``output`` is
    shown as an *editable answer*, not as persuasion.
    """

    question: str
    output: dict[str, Any]
    input: dict[str, Any] | None = None
    summary: str | None = None

    def to_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"question": self.question, "output": self.output}
        if self.input is not None:
            out["input"] = self.input
        if self.summary is not None:
            out["summary"] = self.summary
        return out

DecisionRequest dataclass

The gateway-proof HITL seam (spec §4). See module docstring.

agent_display_name is the agent's display name. agent_name is its deprecated alias, removed in coalex 2.0: it always holds the same value, is serialized beside it, and passing it instead of agent_display_name warns once. When both are passed, agent_display_name wins, so dataclasses.replace(dr, agent_display_name=...) changes both.

Source code in coalex/decision_request.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
@dataclasses.dataclass(frozen=True)
class DecisionRequest:
    """The gateway-proof HITL seam (spec §4). See module docstring.

    ``agent_display_name`` is the agent's display name. ``agent_name`` is its deprecated alias,
    removed in coalex 2.0: it always holds the same value, is serialized beside it, and passing it
    instead of ``agent_display_name`` warns once. When both are passed, ``agent_display_name`` wins,
    so ``dataclasses.replace(dr, agent_display_name=...)`` changes both.
    """

    id: str
    task_state: TaskState
    context: ContextPayload
    question_schema: QuestionSchema
    resolver: ResolverPolicy
    account_id: str | None = None
    agent_id: str | None = None
    agent_name: str | None = None
    # The governed call this request is about, when the producer named one. The gateway path always
    # does; an SDK ``evaluate()`` escalation is about an output rather than a named tool and leaves
    # it None. Carried so the audit rail can say WHICH call paused instead of repeating
    # the question — and absent rather than guessed, because a rail that invents a tool name is
    # worse than one that admits it does not have it.
    tool_name: str | None = None
    request_id: str | None = None
    resume_hook: ResumeHook | None = None
    result: ElicitationResult | None = None
    # When this stops being worth waiting for. None = no deadline. Consumers show it so a
    # reviewer can see how long they have, rather than discovering the answer when it disappears.
    expires_at: str | None = None
    created_at: str | None = None
    updated_at: str | None = None
    # Last, so a positional construction that stops before it keeps its meaning.
    agent_display_name: str | None = None

    def __post_init__(self) -> None:
        if self.agent_display_name is None and self.agent_name is not None:
            _warn_once(_AGENT_NAME_DEPRECATION)
            object.__setattr__(self, "agent_display_name", self.agent_name)
        else:
            object.__setattr__(self, "agent_name", self.agent_display_name)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to the wire JSON the Console (and future Gateway) consume.

        Keys are snake_case and mirror the TypeScript SDK's ``serializeDecisionRequest``
        exactly, so the seam is one shared contract across both SDKs.
        """
        out: dict[str, Any] = {
            "id": self.id,
            "task_state": self.task_state.value,
            "context": self.context.to_dict(),
            "question_schema": dict(self.question_schema),
            "resolver": self.resolver.to_spec(),
        }
        if self.account_id is not None:
            out["account_id"] = self.account_id
        if self.agent_id is not None:
            out["agent_id"] = self.agent_id
        if self.agent_display_name is not None:
            out["agent_display_name"] = self.agent_display_name
            out["agent_name"] = self.agent_display_name
        if self.tool_name is not None:
            out["tool_name"] = self.tool_name
        if self.request_id is not None:
            out["request_id"] = self.request_id
        if self.resume_hook is not None:
            out["resume_hook"] = self.resume_hook.to_dict()
        if self.result is not None:
            out["result"] = self.result.to_dict()
        if self.expires_at is not None:
            out["expires_at"] = self.expires_at
        if self.created_at is not None:
            out["created_at"] = self.created_at
        if self.updated_at is not None:
            out["updated_at"] = self.updated_at
        return out
Methods:
to_dict 
to_dict() -> dict[str, Any]

Serialize to the wire JSON the Console (and future Gateway) consume.

Keys are snake_case and mirror the TypeScript SDK's serializeDecisionRequest exactly, so the seam is one shared contract across both SDKs.

Source code in coalex/decision_request.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def to_dict(self) -> dict[str, Any]:
    """Serialize to the wire JSON the Console (and future Gateway) consume.

    Keys are snake_case and mirror the TypeScript SDK's ``serializeDecisionRequest``
    exactly, so the seam is one shared contract across both SDKs.
    """
    out: dict[str, Any] = {
        "id": self.id,
        "task_state": self.task_state.value,
        "context": self.context.to_dict(),
        "question_schema": dict(self.question_schema),
        "resolver": self.resolver.to_spec(),
    }
    if self.account_id is not None:
        out["account_id"] = self.account_id
    if self.agent_id is not None:
        out["agent_id"] = self.agent_id
    if self.agent_display_name is not None:
        out["agent_display_name"] = self.agent_display_name
        out["agent_name"] = self.agent_display_name
    if self.tool_name is not None:
        out["tool_name"] = self.tool_name
    if self.request_id is not None:
        out["request_id"] = self.request_id
    if self.resume_hook is not None:
        out["resume_hook"] = self.resume_hook.to_dict()
    if self.result is not None:
        out["result"] = self.result.to_dict()
    if self.expires_at is not None:
        out["expires_at"] = self.expires_at
    if self.created_at is not None:
        out["created_at"] = self.created_at
    if self.updated_at is not None:
        out["updated_at"] = self.updated_at
    return out

ElicitationAction

Bases: StrEnum

Elicitation result action for a resolved task (spec §5).

accept = approved (optionally with content = corrections); decline = rejected (+ reason, flows back to the agent as deny-with-instruction); cancel = expired / timed out.

Source code in coalex/decision_request.py 
116
117
118
119
120
121
122
123
124
125
126
class ElicitationAction(StrEnum):
    """Elicitation result action for a resolved task (spec §5).

    ``accept`` = approved (optionally *with content* = corrections); ``decline`` =
    rejected (+ reason, flows back to the agent as deny-with-instruction); ``cancel`` =
    expired / timed out.
    """

    ACCEPT = "accept"
    DECLINE = "decline"
    CANCEL = "cancel"

ElicitationResult dataclass

The resolution of a task (spec §5, §7.2).

Carries the elicitation action plus the DORA who-did-what trail: reviewer (SSO identity), reason, and timestamp. content holds corrections for an accept-with-content; reason holds the decline instruction returned to the agent.

Source code in coalex/decision_request.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
@dataclasses.dataclass(frozen=True)
class ElicitationResult:
    """The resolution of a task (spec §5, §7.2).

    Carries the elicitation ``action`` plus the DORA who-did-what trail: ``reviewer``
    (SSO identity), reason, and timestamp. ``content`` holds corrections for an
    accept-with-content; ``reason`` holds the decline instruction returned to the agent.
    """

    action: ElicitationAction
    content: dict[str, Any] | None = None
    reason: str | None = None
    reviewer: str | None = None
    reviewer_name: str | None = None
    reviewer_email: str | None = None
    resolved_at: str | None = None

    def to_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"action": self.action.value}
        if self.content is not None:
            out["content"] = self.content
        if self.reason is not None:
            out["reason"] = self.reason
        if self.reviewer is not None:
            out["reviewer"] = self.reviewer
        if self.reviewer_name is not None:
            out["reviewer_name"] = self.reviewer_name
        if self.reviewer_email is not None:
            out["reviewer_email"] = self.reviewer_email
        if self.resolved_at is not None:
            out["resolved_at"] = self.resolved_at
        return out

IdentityComposer

v1 (spec §6): the card shows what the producer sent, minus what §6 keeps out.

Deliberately does nothing interesting. Its whole value is that it is a SEAM — swapping it for a composer that renders a long document as a document, or annotates a recipient outside the organisation, changes the card with no change to the card's code.

The one thing it does do is drop the forbidden keys. That is not new behaviour for the Console, which has always filtered them at render and still does; it is new for everyone else, because until now an agent's reasoning or risk_score reached the wire and only one client knew to look away.

Source code in coalex/decision_request.py 
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
class IdentityComposer:
    """v1 (spec §6): the card shows what the producer sent, minus what §6 keeps out.

    Deliberately does nothing interesting. Its whole value is that it is a SEAM — swapping it for a
    composer that renders a long document as a document, or annotates a recipient outside the
    organisation, changes the card with no change to the card's code.

    The one thing it does do is drop the forbidden keys. That is not new behaviour for the Console,
    which has always filtered them at render and still does; it is new for everyone else,
    because until now an agent's `reasoning` or `risk_score` reached the wire and only one client
    knew to look away.
    """

    def compose(self, escalation: Mapping[str, Any] | Any) -> ComposedQuestion:
        raw_output = _get(escalation, "output", {}) or {}
        scrubbed = scrub_forbidden(raw_output)
        output = dict(scrubbed) if isinstance(scrubbed, Mapping) else {}
        raw_metrics = _get(escalation, "metrics", None)
        return ComposedQuestion(
            question=DEFAULT_QUESTION,
            # A human-readable reason the producer curated for the card, carried on the
            # escalation's ``metadata.reason``. ONLY this field is surfaced from metadata — never the
            # risk tier, governed flag or risk score (§6). Absent → no summary.
            summary=_reason_from_metadata(_get(escalation, "metadata", None)),
            output=output,
            # Derived from the FILTERED output, so a forbidden key cannot come back as a field.
            schema=_question_schema_from_output(output, raw_metrics if isinstance(raw_metrics, Mapping) else None),
        )

LegibleComposer

Marks the fields a single-line input cannot show, so the card can render them as documents.

This is the composer the seam was built for, and the defect is real: a Connect IQ approval carries a ~2,000-character markdown brief — headings, bullet lists, client and campaign facts. Its JSON type is string, so the card falls through to its last branch and renders it in a one-line <Input>. The approver is then asked to tick "Checked" against facts they were never shown. An approval nobody can read is not governance; it is a rubber stamp that leaves an audit trail.

It DECORATES another composer rather than replacing it, because legibility is orthogonal to what the question says: LegibleComposer(GatewayComposer()) keeps the gateway's "Approve calling 'wire_funds'…" question and still makes a long argument readable. Stacking is the point — an all-or-nothing composer would force every deployment to reimplement the others.

It adds no content. §6 is not relaxed by making things legible: this changes how a value is SHOWN, never which values exist, and never adds a word of its own to the card.

Source code in coalex/decision_request.py 
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
class LegibleComposer:
    """Marks the fields a single-line input cannot show, so the card can render them as documents.

    This is the composer the seam was built for, and the defect is real: a Connect IQ approval
    carries a ~2,000-character markdown ``brief`` — headings, bullet lists, client and campaign
    facts. Its JSON type is ``string``, so the card falls through to its last branch and renders it
    in a one-line ``<Input>``. The approver is then asked to tick "Checked" against facts they were
    never shown. An approval nobody can read is not governance; it is a rubber stamp that leaves an
    audit trail.

    It DECORATES another composer rather than replacing it, because legibility is orthogonal to
    what the question says: ``LegibleComposer(GatewayComposer())`` keeps the gateway's "Approve
    calling 'wire_funds'…" question and still makes a long argument readable. Stacking is the
    point — an all-or-nothing composer would force every deployment to reimplement the others.

    It adds no content. §6 is not relaxed by making things legible: this changes how a value is
    SHOWN, never which values exist, and never adds a word of its own to the card.
    """

    def __init__(self, inner: QuestionComposer | None = None) -> None:
        self._inner = inner if inner is not None else IdentityComposer()

    def compose(self, escalation: Mapping[str, Any] | Any) -> ComposedQuestion:
        composed = self._inner.compose(escalation)
        annotated: dict[str, QuestionField] = {}
        marked = False
        for name, field in composed.schema["properties"].items():
            # Only strings: an array or an object already has a widget that shows its shape. And
            # the value is read from the COMPOSED output, never the raw escalation, so a field the
            # inner composer dropped or rewrote is judged on what the card will actually receive.
            if field.get("type") == "string" and _is_document(composed.output.get(name)):
                document = dict(field)
                document["coalex.format"] = "document"
                annotated[name] = cast(QuestionField, document)
                marked = True
            else:
                annotated[name] = field
        if not marked:
            return composed
        schema: QuestionSchema = {
            "type": "object",
            "properties": annotated,
            # ``.get`` because a composer is free to hand back a schema without it — the SDK's own
            # tests use one. Annotating a field must never be the thing that raises.
            "required": composed.schema.get("required", []),
        }
        return dataclasses.replace(composed, schema=schema)

QuestionComposer

Bases: Protocol

Turns an escalation into what the approver reads (spec §6).

The indirection exists so the card never renders the raw payload. A producer's output is shaped for a machine — a 2,000-character markdown brief is a string, and the card, being schema-driven, renders it in a single-line input the approver cannot read. Deciding that such a value is a document and not a text field is this stage's job; there is nowhere else it belongs, because the schema is the producer's and the card is generic.

Composers must not add what §6 keeps out. Making the facts LEGIBLE is the job; explaining them is not, and an agent's reasoning, confidence or risk score must not reach the card through a composed summary any more than through a field.

Source code in coalex/decision_request.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class QuestionComposer(Protocol):
    """Turns an escalation into what the approver reads (spec §6).

    The indirection exists so the card never renders the raw payload. A producer's output is shaped
    for a machine — a 2,000-character markdown brief is a `string`, and the card, being
    schema-driven, renders it in a single-line input the approver cannot read. Deciding that such a
    value is a document and not a text field is this stage's job; there is nowhere else it belongs,
    because the schema is the producer's and the card is generic.

    Composers must not add what §6 keeps out. Making the facts LEGIBLE is the job; explaining them
    is not, and an agent's reasoning, confidence or risk score must not reach the card through a
    composed summary any more than through a field.
    """

    def compose(self, escalation: Mapping[str, Any] | Any) -> ComposedQuestion: ...

ResolverPolicy dataclass

Who may resolve this request (spec §4.3): self or role:<name>.

Source code in coalex/decision_request.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@dataclasses.dataclass(frozen=True)
class ResolverPolicy:
    """Who may resolve this request (spec §4.3): ``self`` or ``role:<name>``."""

    kind: Literal["self", "role"]
    role: str | None = None

    @classmethod
    def parse(cls, spec: str) -> ResolverPolicy:
        """Parse the ``self`` | ``role:<name>`` config string into a policy."""
        if spec == "self":
            return cls(kind="self")
        if spec.startswith("role:"):
            role = spec[len("role:") :].strip()
            if not role:
                raise ValueError("role resolver requires a name, e.g. 'role:manager'")
            return cls(kind="role", role=role)
        raise ValueError(f"resolver must be 'self' or 'role:<name>', got {spec!r}")

    def to_spec(self) -> str:
        """Serialize back to the ``self`` | ``role:<name>`` wire string."""
        if self.kind == "role":
            return f"role:{self.role}"
        return "self"
Methods:
parse classmethod 
parse(spec: str) -> ResolverPolicy

Parse the self | role:<name> config string into a policy.

Source code in coalex/decision_request.py 
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def parse(cls, spec: str) -> ResolverPolicy:
    """Parse the ``self`` | ``role:<name>`` config string into a policy."""
    if spec == "self":
        return cls(kind="self")
    if spec.startswith("role:"):
        role = spec[len("role:") :].strip()
        if not role:
            raise ValueError("role resolver requires a name, e.g. 'role:manager'")
        return cls(kind="role", role=role)
    raise ValueError(f"resolver must be 'self' or 'role:<name>', got {spec!r}")
to_spec 
to_spec() -> str

Serialize back to the self | role:<name> wire string.

Source code in coalex/decision_request.py 
184
185
186
187
188
def to_spec(self) -> str:
    """Serialize back to the ``self`` | ``role:<name>`` wire string."""
    if self.kind == "role":
        return f"role:{self.role}"
    return "self"

ResumeHook dataclass

How the producer resumes after resolution (spec §4.4).

v1 (SDK path): the LangGraph checkpoint-resume callback promoted to a first-class field — the agent registers its resume url (and/or checkpoint_id) when it escalates. Future (Gateway path): completing/failing the held MCP task_id.

Source code in coalex/decision_request.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@dataclasses.dataclass(frozen=True)
class ResumeHook:
    """How the producer resumes after resolution (spec §4.4).

    v1 (SDK path): the LangGraph checkpoint-resume callback promoted to a first-class
    field — the agent registers its resume ``url`` (and/or ``checkpoint_id``) when it
    escalates. Future (Gateway path): completing/failing the held MCP ``task_id``.
    """

    kind: Literal["checkpoint", "mcp_task"] = "checkpoint"
    url: str | None = None
    checkpoint_id: str | None = None
    task_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"kind": self.kind}
        if self.url is not None:
            out["url"] = self.url
        if self.checkpoint_id is not None:
            out["checkpoint_id"] = self.checkpoint_id
        if self.task_id is not None:
            out["task_id"] = self.task_id
        return out

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> ResumeHook:
        """Rebuild a ResumeHook from its ``to_dict()`` wire shape (spec §4.4).

        The inverse of ``to_dict`` — used when a persisted escalation carries a stored hook
        (``escalations.resume_hook``) that must become a ``DecisionRequest``. An
        unrecognized ``kind`` falls back to ``checkpoint`` rather than raising, so a forward
        Gateway value can never break the SDK read path.
        """
        kind = data.get("kind")
        return cls(
            kind=kind if kind in ("checkpoint", "mcp_task") else "checkpoint",
            url=data.get("url"),
            checkpoint_id=data.get("checkpoint_id"),
            task_id=data.get("task_id"),
        )
Methods:
from_dict classmethod 
from_dict(data: Mapping[str, Any]) -> ResumeHook

Rebuild a ResumeHook from its to_dict() wire shape (spec §4.4).

The inverse of to_dict — used when a persisted escalation carries a stored hook (escalations.resume_hook) that must become a DecisionRequest. An unrecognized kind falls back to checkpoint rather than raising, so a forward Gateway value can never break the SDK read path.

Source code in coalex/decision_request.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ResumeHook:
    """Rebuild a ResumeHook from its ``to_dict()`` wire shape (spec §4.4).

    The inverse of ``to_dict`` — used when a persisted escalation carries a stored hook
    (``escalations.resume_hook``) that must become a ``DecisionRequest``. An
    unrecognized ``kind`` falls back to ``checkpoint`` rather than raising, so a forward
    Gateway value can never break the SDK read path.
    """
    kind = data.get("kind")
    return cls(
        kind=kind if kind in ("checkpoint", "mcp_task") else "checkpoint",
        url=data.get("url"),
        checkpoint_id=data.get("checkpoint_id"),
        task_id=data.get("task_id"),
    )

TaskState

Bases: StrEnum

MCP task states carried by a Decision Request (spec §5).

working (agent in-flight, pre-escalation) and failed (technical error / future Gateway path) have no current escalation-status equivalent; they exist so the enum is complete for the shared Console+Gateway backbone.

Source code in coalex/decision_request.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
class TaskState(StrEnum):
    """MCP task states carried by a Decision Request (spec §5).

    ``working`` (agent in-flight, pre-escalation) and ``failed`` (technical error /
    future Gateway path) have no current escalation-status equivalent; they exist so
    the enum is complete for the shared Console+Gateway backbone.
    """

    WORKING = "working"
    INPUT_REQUIRED = "input_required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

EvaluationDecision dataclass

Result of an evaluate() call.

Source code in coalex/evaluate.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclasses.dataclass(frozen=True)
class EvaluationDecision:
    """Result of an evaluate() call."""

    # "auto_approved" | "escalated". `rejected` was listed here and the server has never
    # returned it: /evaluate chooses between those two and nothing else. A human
    # rejecting an escalation is a different event, recorded on the escalation.
    status: str
    risk_score: float  # 0.0-1.0
    escalation_id: str | None  # present when status == "escalated"
    # The same outcome in MCP vocabulary: "escalated" is "input_required". None against an
    # older platform that does not send it yet.
    task_state: str | None = None

GuardResult dataclass

Outcome of :func:guard_tool_call.

decision is the final branch: allow (tool ran, result holds its return value), escalate (paused pending a human — escalation_id is the Decision Request), or block (denied; tool never ran).

Source code in coalex/gateway.py 
70
71
72
73
74
75
76
77
78
79
80
81
@dataclasses.dataclass(frozen=True)
class GuardResult:
    """Outcome of :func:`guard_tool_call`.

    ``decision`` is the *final* branch: ``allow`` (tool ran, ``result`` holds its return value),
    ``escalate`` (paused pending a human — ``escalation_id`` is the Decision Request), or ``block``
    (denied; tool never ran).
    """

    decision: PolicyDecision
    result: Any = None
    escalation_id: str | None = None

AgentInventory dataclass

What an agent declares about itself.

An agent of a multi-agent system declares itself once per agent: the umbrella with the endpoint the Console talks to and its resume hook, and each child with parent and no endpoint, because a child is resumed through its nearest ancestor that has one.

Parameters:

Name Type Description Default
endpoint str | None

Where a turn is POSTed — the URL the Console will talk to. Optional: a child agent that is only ever reached through its parent declares none, and the Console does not offer it for a session. First, so a positional AgentInventory("https://...") keeps working.

 None
tools Sequence[str]

The names of the tools this agent exposes. Names only: whether a tool is governed is the org's call, not the agent's.

 ()
resume_url str | None

Where a paused run is resumed, if the agent supports being paused.

 None
name str | None

Deprecated alias of agent_id, removed in coalex 2.0. It warns once and is sent as agent_id.

 None
display_name str | None

The agent's title, shown on a new chat in the Console.

 None
description str | None

How the agent introduces itself on a new chat, shown as its first message.

 None
suggested_prompts Sequence[str] | None

What an operator can ask it, shown as buttons that send that text. The Console serves every customer's agents, so it shows these as given and invents nothing when they are absent: an agent without them is shown by its name alone.

 None
agent_id str | None

The agent's id, the one key the platform files its traces, policy and health under. Omitted, it is the agent_id of the active coalex_context(); outside one, register()'s service_name when the caller passed one (deprecated, removed in coalex 2.0). The default service name is never an id: such a declaration is refused and logged.

 None
parent str | None

The agent this one proposes as its parent. It governs nothing until the org confirms it in the dashboard; a re-declaration without it keeps the confirmed parent.

 None

An id is checked before anything is sent, by the rule the platform applies: it must not be empty, start or end with whitespace, or be longer than 128 characters. Any character, . included, is allowed, and its shape means nothing: a parent is only ever parent.

Source code in coalex/inventory.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@dataclasses.dataclass(frozen=True)
class AgentInventory:
    """What an agent declares about itself.

    An agent of a multi-agent system declares itself once per agent: the umbrella with the
    endpoint the Console talks to and its resume hook, and each child with ``parent`` and no
    endpoint, because a child is resumed through its nearest ancestor that has one.

    Args:
        endpoint: Where a turn is POSTed — the URL the Console will talk to. Optional: a child
            agent that is only ever reached through its parent declares none, and the Console does
            not offer it for a session. First, so a positional ``AgentInventory("https://...")``
            keeps working.
        tools: The names of the tools this agent exposes. Names only: whether a tool is governed
            is the org's call, not the agent's.
        resume_url: Where a paused run is resumed, if the agent supports being paused.
        name: Deprecated alias of ``agent_id``, removed in coalex 2.0. It warns once and is sent as
            ``agent_id``.
        display_name: The agent's title, shown on a new chat in the Console.
        description: How the agent introduces itself on a new chat, shown as its first message.
        suggested_prompts: What an operator can ask it, shown as buttons that send that text.
            The Console serves every customer's agents, so it shows these as given and invents
            nothing when they are absent: an agent without them is shown by its name alone.
        agent_id: The agent's id, the one key the platform files its traces, policy and health
            under. Omitted, it is the ``agent_id`` of the active ``coalex_context()``; outside one,
            ``register()``'s ``service_name`` when the caller passed one (deprecated, removed in
            coalex 2.0). The default service name is never an id: such a declaration is refused
            and logged.
        parent: The agent this one proposes as its parent. It governs nothing until the org
            confirms it in the dashboard; a re-declaration without it keeps the confirmed parent.

    An id is checked before anything is sent, by the rule the platform applies: it must not be
    empty, start or end with whitespace, or be longer than 128 characters. Any character, ``.``
    included, is allowed, and its shape means nothing: a parent is only ever ``parent``.
    """

    endpoint: str | None = None
    tools: Sequence[str] = ()
    resume_url: str | None = None
    name: str | None = None
    display_name: str | None = None
    description: str | None = None
    suggested_prompts: Sequence[str] | None = None
    agent_id: str | None = None
    parent: str | None = None

    def __post_init__(self) -> None:
        if self.name is None:
            return
        if self.agent_id is not None and self.agent_id != self.name:
            raise ValueError("pass agent_id or its deprecated alias name, not both")
        _warn_once(_NAME_DEPRECATION)
        object.__setattr__(self, "agent_id", self.name)

    def payload(self, *, sdk: str, default_name: str | None = None) -> dict[str, Any]:
        """The POST body. ``default_name`` is the id used when the declaration names none."""
        agent_id = self.agent_id or default_name
        if agent_id is None:
            raise ValueError("an agent declaration needs an agent_id")
        body: dict[str, Any] = {
            "agent_id": agent_id,
            "tools": list(self.tools),
            "sdk": sdk,
        }
        # Only what was declared: the platform answers 422 to a key it does not know, and an absent
        # `parent` keeps the parent the org confirmed.
        if self.endpoint is not None:
            body["endpoint"] = self.endpoint
        if self.parent is not None:
            body["parent"] = self.parent
        if self.resume_url:
            body["resume_hook"] = {"kind": "checkpoint", "url": self.resume_url}
        if self.display_name is not None:
            body["display_name"] = self.display_name
        if self.description is not None:
            body["description"] = self.description
        if self.suggested_prompts is not None:
            body["suggested_prompts"] = list(self.suggested_prompts)
        return body
Methods:
payload 
payload(
    *, sdk: str, default_name: str | None = None
) -> dict[str, Any]

The POST body. default_name is the id used when the declaration names none.

Source code in coalex/inventory.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def payload(self, *, sdk: str, default_name: str | None = None) -> dict[str, Any]:
    """The POST body. ``default_name`` is the id used when the declaration names none."""
    agent_id = self.agent_id or default_name
    if agent_id is None:
        raise ValueError("an agent declaration needs an agent_id")
    body: dict[str, Any] = {
        "agent_id": agent_id,
        "tools": list(self.tools),
        "sdk": sdk,
    }
    # Only what was declared: the platform answers 422 to a key it does not know, and an absent
    # `parent` keeps the parent the org confirmed.
    if self.endpoint is not None:
        body["endpoint"] = self.endpoint
    if self.parent is not None:
        body["parent"] = self.parent
    if self.resume_url:
        body["resume_hook"] = {"kind": "checkpoint", "url": self.resume_url}
    if self.display_name is not None:
        body["display_name"] = self.display_name
    if self.description is not None:
        body["description"] = self.description
    if self.suggested_prompts is not None:
        body["suggested_prompts"] = list(self.suggested_prompts)
    return body

PromptVersion dataclass

A single prompt version returned by the Prompt Vault.

Source code in coalex/prompts.py 
30
31
32
33
34
35
36
37
38
39
40
41
@dataclasses.dataclass(frozen=True)
class PromptVersion:
    """A single prompt version returned by the Prompt Vault."""

    id: str
    agent_id: str
    name: str
    prompt_type: str  # "system" | "guardrails"
    version: int
    content: str
    status: str  # "draft" | "production" — the only two the Prompt Vault has
    metadata: dict[str, Any]

ResolutionResult dataclass

Result of a resolve() call.

Source code in coalex/resolve.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@dataclasses.dataclass(frozen=True)
class ResolutionResult:
    """Result of a resolve() call."""

    escalation_id: str
    status: str  # "approved" | "rejected" | "corrected" | "expired"
    metrics: list[MetricResult]
    resolved_at: str
    resolved_by: str | None = None
    reviewer_name: str | None = None
    reviewer_email: str | None = None
    # The MCP reading of the same resolution. None against an older platform that does not
    # send them yet — the SDK reports what it was told rather than inventing the fields locally, so a
    # caller can tell "this deployment is older" from "this decision has no action".
    task_state: str | None = None
    result_action: str | None = None

Functions:

declare_agent 

declare_agent(
    *,
    agent_id: str,
    display_name: str | None = None,
    metadata: dict[str, Any] | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> AgentDeclaration

Register or declare an agent before traces arrive.

This is the building block for lazy agent registration. When the Prompt Vault's set_prompt() is implemented, it will call this internally to ensure the agent exists.

Parameters:

Name Type Description Default
agent_id str

Unique agent identifier (the merge key).

 required
display_name str | None

Optional human-friendly name.

 None
metadata dict[str, Any] | None

Optional metadata (accepted but not persisted yet).

 None
_config CoalexConfig | None

Override config (testing only).

 None
_api_key str | None

Override API key (testing only).

 None
Source code in coalex/agents.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def declare_agent(
    *,
    agent_id: str,
    display_name: str | None = None,
    metadata: dict[str, Any] | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> AgentDeclaration:
    """Register or declare an agent before traces arrive.

    This is the building block for lazy agent registration.
    When the Prompt Vault's set_prompt() is implemented, it will call
    this internally to ensure the agent exists.

    Args:
        agent_id: Unique agent identifier (the merge key).
        display_name: Optional human-friendly name.
        metadata: Optional metadata (accepted but not persisted yet).
        _config: Override config (testing only).
        _api_key: Override API key (testing only).
    """
    import coalex as _sdk

    cfg = _config or _sdk._config
    api_key = _api_key or _sdk._api_key
    if cfg is None:
        raise RuntimeError("coalex.register() must be called before declare_agent()")

    payload: dict[str, Any] = {"agent_id": agent_id}
    if display_name is not None:
        payload["display_name"] = display_name
    if metadata is not None:
        payload["metadata"] = metadata

    with httpx.Client(timeout=30.0) as client:
        resp = client.post(
            f"{cfg.endpoint}/api/v1/agents",
            json=payload,
            headers=sdk_headers(api_key),
        )
        raise_for_status(resp, "declare_agent()")

    data = resp.json()
    return AgentDeclaration(
        agent_id=data["agent_id"],
        lifecycle=data["lifecycle"],
        created=data["created"],
    )

instrument_anthropic 

instrument_anthropic(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument Anthropic Claude specifically.

Source code in coalex/auto_instrument.py 
425
426
427
def instrument_anthropic(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument Anthropic Claude specifically."""
    return _convenience_instrument("anthropic", tracer_provider)

instrument_bedrock 

instrument_bedrock(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument AWS Bedrock specifically.

Source code in coalex/auto_instrument.py 
420
421
422
def instrument_bedrock(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument AWS Bedrock specifically."""
    return _convenience_instrument("bedrock", tracer_provider)

instrument_google_genai 

instrument_google_genai(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument Google GenAI (google-genai SDK) specifically.

Source code in coalex/auto_instrument.py 
410
411
412
def instrument_google_genai(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument Google GenAI (google-genai SDK) specifically."""
    return _convenience_instrument("google_genai", tracer_provider)

instrument_langchain 

instrument_langchain(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument LangChain specifically.

Source code in coalex/auto_instrument.py 
405
406
407
def instrument_langchain(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument LangChain specifically."""
    return _convenience_instrument("langchain", tracer_provider)

instrument_llamaindex 

instrument_llamaindex(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument LlamaIndex specifically.

Source code in coalex/auto_instrument.py 
400
401
402
def instrument_llamaindex(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument LlamaIndex specifically."""
    return _convenience_instrument("llamaindex", tracer_provider)

instrument_openai 

instrument_openai(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument OpenAI SDK specifically.

Source code in coalex/auto_instrument.py 
395
396
397
def instrument_openai(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument OpenAI SDK specifically."""
    return _convenience_instrument("openai", tracer_provider)

instrument_vertexai 

instrument_vertexai(
    tracer_provider: TracerProvider | None = None,
) -> bool

Instrument Google VertexAI specifically.

Source code in coalex/auto_instrument.py 
415
416
417
def instrument_vertexai(tracer_provider: TracerProvider | None = None) -> bool:
    """Instrument Google VertexAI specifically."""
    return _convenience_instrument("vertexai", tracer_provider)

verify_console_call 

verify_console_call(
    token: str | None,
    *,
    audience: str,
    public_keys: Mapping[str, str] | str,
    expected_account: str | None = None,
    leeway_seconds: int = DEFAULT_LEEWAY_SECONDS,
    now: float | None = None,
) -> ConsoleCaller

Verify a Console-signed call, returning who it is for. Raises :class:ConsoleAuthError.

Parameters:

Name Type Description Default
token str | None

The bearer token from the request, without the Bearer prefix.

 required
audience str

This agent's own name. A token minted for another agent is refused.

 required
expected_account str | None

The account this agent belongs to. A token the Console minted for any other account is refused, even when it names this agent. Agent names are unique only within an account and any organisation can declare its own agent endpoint, so the name alone does not keep one organisation's Console out of another organisation's agent of the same name; the account does. Omitting it keeps the old behaviour (the name alone) and warns with :class:MissingExpectedAccountWarning: it becomes required in coalex 2.0. An empty string refuses every token, so an unset variable read as "" does not turn the check off.

 None
public_keys Mapping[str, str] | str

{kid: pem}, or a single PEM when no key id is in use. More than one key can be valid at once so a signing key can be rotated without redeploying every agent.

 required
leeway_seconds int

Clock skew tolerance.

 DEFAULT_LEEWAY_SECONDS
now float | None

Override the clock, for tests.

 None

The order matters: the signature is checked before any claim is read. Claims from an unverified token are attacker-controlled strings, and reading them first — even to pick a key — is how verification gets subverted. Only kid, from the header, selects the key, and an unknown kid is a refusal rather than a fallback to some default key.

Source code in coalex/console_auth.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def verify_console_call(
    token: str | None,
    *,
    audience: str,
    public_keys: Mapping[str, str] | str,
    expected_account: str | None = None,
    leeway_seconds: int = DEFAULT_LEEWAY_SECONDS,
    now: float | None = None,
) -> ConsoleCaller:
    """Verify a Console-signed call, returning who it is for. Raises :class:`ConsoleAuthError`.

    Args:
        token: The bearer token from the request, without the ``Bearer `` prefix.
        audience: This agent's own name. A token minted for another agent is refused.
        expected_account: The account this agent belongs to. A token the Console minted for any other
            account is refused, even when it names this agent. Agent names are unique only within an
            account and any organisation can declare its own agent endpoint, so the name alone does
            not keep one organisation's Console out of another organisation's agent of the same
            name; the account does. Omitting it keeps the old behaviour (the name alone) and warns
            with :class:`MissingExpectedAccountWarning`: it becomes required in coalex 2.0. An empty
            string refuses every token, so an unset variable read as ``""`` does not turn the check
            off.
        public_keys: ``{kid: pem}``, or a single PEM when no key id is in use. More than one key can
            be valid at once so a signing key can be rotated without redeploying every agent.
        leeway_seconds: Clock skew tolerance.
        now: Override the clock, for tests.

    The order matters: the signature is checked **before** any claim is read. Claims from an
    unverified token are attacker-controlled strings, and reading them first — even to pick a key —
    is how verification gets subverted. Only ``kid``, from the header, selects the key, and an
    unknown ``kid`` is a refusal rather than a fallback to some default key.
    """
    if not token:
        raise ConsoleAuthError("no token")

    parts = token.split(".")
    if len(parts) != 3:
        raise ConsoleAuthError("not a compact JWS")
    header_b64, payload_b64, signature_b64 = parts

    try:
        header = json.loads(_b64url_decode(header_b64))
    except (ValueError, TypeError) as exc:
        raise ConsoleAuthError(f"header is not JSON: {exc}") from exc
    if not isinstance(header, dict):
        raise ConsoleAuthError("header is not an object")

    if header.get("alg") != _ALGORITHM:
        # Pinned, never read from the token. `alg: none` is the canonical way this goes wrong.
        raise ConsoleAuthError(f"algorithm {header.get('alg')!r} is not accepted")

    if isinstance(public_keys, str):
        pem = public_keys
    else:
        kid = header.get("kid")
        if not isinstance(kid, str) or kid not in public_keys:
            # No default key: an unknown kid must fail, not silently try another key.
            raise ConsoleAuthError(f"unknown key id {kid!r}")
        pem = public_keys[kid]

    key = _load_public_key(pem)
    signing_input = f"{header_b64}.{payload_b64}".encode()
    try:
        key.verify(_b64url_decode(signature_b64), signing_input)
    except ConsoleAuthError:
        raise
    except Exception as exc:  # InvalidSignature and anything else are one answer
        raise ConsoleAuthError(f"signature does not verify: {exc}") from exc

    # Only now is anything in the payload worth reading.
    try:
        claims = json.loads(_b64url_decode(payload_b64))
    except (ValueError, TypeError) as exc:
        raise ConsoleAuthError(f"payload is not JSON: {exc}") from exc
    if not isinstance(claims, dict):
        raise ConsoleAuthError("payload is not an object")

    if claims.get("aud") != audience:
        raise ConsoleAuthError(f"token is for {claims.get('aud')!r}, not {audience!r}")

    account_id = claims.get("sub")
    if not isinstance(account_id, str) or not account_id:
        raise ConsoleAuthError("no subject")
    if expected_account is None:
        warnings.warn(
            "verify_console_call() was called without expected_account: it checks that the token was "
            "minted for this agent's name but not for this agent's account, so another organisation's "
            "Console can reach an agent of the same name. Pass expected_account=<the account this "
            "agent belongs to>. expected_account becomes required in coalex 2.0.",
            MissingExpectedAccountWarning,
            stacklevel=2,
        )
    elif account_id != expected_account:
        raise ConsoleAuthError(f"token is for account {account_id!r}, not {expected_account!r}")

    clock = time.time() if now is None else now
    exp = claims.get("exp")
    if not isinstance(exp, (int, float)):
        # An unexpiring token is not a short-lived one, whatever it claims to be.
        raise ConsoleAuthError("no expiry")
    if clock > exp + leeway_seconds:
        raise ConsoleAuthError("expired")

    iat = claims.get("iat")
    if isinstance(iat, (int, float)) and iat > clock + leeway_seconds:
        # Issued in the future: either a badly skewed clock or a forged claim.
        raise ConsoleAuthError("issued in the future")

    return ConsoleCaller(
        account_id=account_id,
        agent_id=audience,
        user_id=claims.get("user_id") if isinstance(claims.get("user_id"), str) else None,
        user_email=claims.get("user_email") if isinstance(claims.get("user_email"), str) else None,
        claims=claims,
    )

agent_node 

agent_node(
    agent_id: str, fn: _F, *, version: str | None = None
) -> _F

Wrap a LangGraph node function so everything it does belongs to agent_id.

Wrap the node function, not the graph's invoke: LangGraph re-executes the node from the top on every resume (Command(resume=...)), so the context opens again for the resumed work, and the node's checkpoint namespace is unchanged. The context carries the enclosing request_id and run_id, names the enclosing agent as coalex.parent_agent_id, and a pause (interrupt()) leaves its span OK.

Works for sync and async node functions, and keeps the function's signature, which LangGraph inspects to decide whether to pass config::

graph.add_node("w1_rfq", coalex.agent_node("rfq-worker", rfq_worker, version="1.4.0"))
Source code in coalex/context.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def agent_node(agent_id: str, fn: _F, *, version: str | None = None) -> _F:
    """Wrap a LangGraph node function so everything it does belongs to ``agent_id``.

    Wrap the **node function**, not the graph's ``invoke``: LangGraph re-executes the node from the
    top on every resume (``Command(resume=...)``), so the context opens again for the resumed work,
    and the node's checkpoint namespace is unchanged. The context carries the enclosing
    ``request_id`` and ``run_id``, names the enclosing agent as ``coalex.parent_agent_id``, and a
    pause (``interrupt()``) leaves its span OK.

    Works for sync and async node functions, and keeps the function's signature, which LangGraph
    inspects to decide whether to pass ``config``::

        graph.add_node("w1_rfq", coalex.agent_node("rfq-worker", rfq_worker, version="1.4.0"))
    """
    if version is None:
        _warn_missing_version(agent_id, stacklevel=3)

    if inspect.iscoroutinefunction(fn):

        @functools.wraps(fn)
        async def async_node(*args: Any, **kwargs: Any) -> Any:
            with coalex_context(agent_id=agent_id, version=version):
                return await fn(*args, **kwargs)

        return cast("_F", async_node)

    @functools.wraps(fn)
    def node(*args: Any, **kwargs: Any) -> Any:
        with coalex_context(agent_id=agent_id, version=version):
            return fn(*args, **kwargs)

    return cast("_F", node)

coalex_context 

coalex_context(
    *,
    agent_id: str,
    request_id: str | None = None,
    version: str | None = None,
    prompt_version: int | None = None,
    prompt_type: str | None = None,
    run_id: str | None = None,
    replay_id: str | None = None,
    case_id: str | None = None,
    case_timestamp: datetime | None = None,
) -> Generator[None, None, None]

Create a parent span with Coalex attributes for all child spans.

All spans created inside this context are descendants of a coalex.invocation span that carries the agent metadata. This makes every child span queryable by agent_id, request_id, and version in the Bronze layer.

The coalex.invocation span also carries openinference.span.kind=AGENT, gen_ai.operation.name=invoke_agent and gen_ai.agent.name=<agent_id>, so third-party backends show it as an agent run; its descendants do not.

Context vars are set before the span is created, and CoalexAttributePropagator reads them before the parent span: a framework instrumentor (OpenInference LangChain) parents spans on its own run tree, so a span's parent can belong to an outer agent while this context is active.

Nesting. An inner context wins for every value it is given. Opened inside a context with a different agent, it sets coalex.parent_agent_id to that agent, carries request_id and run_id, and inherits neither version nor the prompt. Inside the same agent it inherits everything and sets no parent. replay_id, case_id and case_timestamp are carried into every nested context, whichever its agent.

Replay. A historical case re-run by a replay passes replay_id, case_id and case_timestamp; every span inside then carries coalex.replay_id, coalex.case_id and coalex.case_timestamp (ISO 8601 with offset), which tie it to its case and keep replay traffic apart from live traffic. Span start times stay the real run time. Outside a replay none of the three is set.

A pause is not an error. An exception whose class, or any base class, is named GraphBubbleUp, GraphInterrupt or ParentCommand (LangGraph's interrupt() and Command(graph=Command.PARENT)) ends the span OK with no exception event and is re-raised. Any other exception ends it ERROR.

Do not wrap a generator that is consumed after the enclosing function returns. A streaming HTTP response is the usual case: the body generator runs once the handler has already returned, so this block exits in a different Context from the one it entered, and the variables cannot be restored. Enter the context inside the generator instead, so its lifetime matches the work it describes::

async def body():
    with coalex_context(agent_id=..., request_id=...):
        async for token in agent.stream(...):
            yield token

Getting this wrong no longer raises — it leaks a context var and logs at debug — but the spans will be attributed to whatever context happens to be current, which is rarely what you meant.

prompt_version / prompt_type let an agent that fetches its prompt once at startup (outside any request trace) attribute each request to a prompt version. get_prompt() only stamps the trace it is called in, so the cache-at-startup pattern needs the version passed here per request. When omitted, any value already set by get_prompt() is left untouched, except inside another agent's context: that prompt is not this agent's.

Parameters:

Name Type Description Default
agent_id str

Identifier for the AI agent.

 required
request_id str | None

The customer's correlation key for one unit of business work. Carried from an enclosing context when omitted.

 None
version str | None

Agent version string — the build this invocation is running. Optional today, and omitting it emits a UserWarning once per agent per process: a trace with no version cannot be attributed to a build, and the dashboard records it against the version unknown. It becomes required in coalex 2.0.

 None
prompt_version int | None

Prompt Vault version driving this invocation.

 None
prompt_type str | None

Prompt type (e.g. "system").

 None
run_id str | None

One execution of an agent graph, across interrupts and resumes. Carried from an enclosing context when omitted. Never derive it from a framework's thread or session id.

 None
replay_id str | None

The replay run this invocation belongs to. Carried from an enclosing context when omitted.

 None
case_id str | None

The historical case being replayed. Carried from an enclosing context when omitted.

 None
case_timestamp datetime | None

When the original case happened. Timezone-aware; a naive datetime raises ValueError, because the attribute must carry an offset. Carried from an enclosing context when omitted.

 None
Source code in coalex/context.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
@contextlib.contextmanager
def coalex_context(
    *,
    agent_id: str,
    request_id: str | None = None,
    version: str | None = None,
    prompt_version: int | None = None,
    prompt_type: str | None = None,
    run_id: str | None = None,
    replay_id: str | None = None,
    case_id: str | None = None,
    case_timestamp: datetime | None = None,
) -> Generator[None, None, None]:
    """Create a parent span with Coalex attributes for all child spans.

    All spans created inside this context are descendants of a
    ``coalex.invocation`` span that carries the agent metadata.
    This makes every child span queryable by agent_id, request_id,
    and version in the Bronze layer.

    The ``coalex.invocation`` span also carries ``openinference.span.kind=AGENT``,
    ``gen_ai.operation.name=invoke_agent`` and ``gen_ai.agent.name=<agent_id>``, so third-party
    backends show it as an agent run; its descendants do not.

    Context vars are set before the span is created, and ``CoalexAttributePropagator`` reads them
    before the parent span: a framework instrumentor (OpenInference LangChain) parents spans on its
    own run tree, so a span's parent can belong to an outer agent while this context is active.

    **Nesting.** An inner context wins for every value it is given. Opened inside a context with a
    *different* agent, it sets ``coalex.parent_agent_id`` to that agent, carries ``request_id`` and
    ``run_id``, and inherits neither ``version`` nor the prompt. Inside the *same* agent it inherits
    everything and sets no parent. ``replay_id``, ``case_id`` and ``case_timestamp`` are carried into
    every nested context, whichever its agent.

    **Replay.** A historical case re-run by a replay passes ``replay_id``, ``case_id`` and
    ``case_timestamp``; every span inside then carries ``coalex.replay_id``, ``coalex.case_id`` and
    ``coalex.case_timestamp`` (ISO 8601 with offset), which tie it to its case and keep replay traffic
    apart from live traffic. Span start times stay the real run time. Outside a replay none of the
    three is set.

    **A pause is not an error.** An exception whose class, or any base class, is named
    ``GraphBubbleUp``, ``GraphInterrupt`` or ``ParentCommand`` (LangGraph's ``interrupt()`` and
    ``Command(graph=Command.PARENT)``) ends the span OK with no exception event and is re-raised.
    Any other exception ends it ERROR.

    **Do not wrap a generator that is consumed after the enclosing function returns.** A streaming
    HTTP response is the usual case: the body generator runs once the handler has already returned,
    so this block exits in a different Context from the one it entered, and the variables cannot be
    restored. Enter the context *inside* the generator instead, so its lifetime matches
    the work it describes::

        async def body():
            with coalex_context(agent_id=..., request_id=...):
                async for token in agent.stream(...):
                    yield token

    Getting this wrong no longer raises — it leaks a context var and logs at debug — but the spans
    will be attributed to whatever context happens to be current, which is rarely what you meant.

    ``prompt_version`` / ``prompt_type`` let an agent that fetches its prompt
    once at startup (outside any request trace) attribute each request to a
    prompt version. ``get_prompt()`` only stamps the trace it is called in, so
    the cache-at-startup pattern needs the version passed here per request.
    When omitted, any value already set by ``get_prompt()`` is left
    untouched, except inside another agent's context: that prompt is not this agent's.

    Args:
        agent_id: Identifier for the AI agent.
        request_id: The customer's correlation key for one unit of business work. Carried from an
            enclosing context when omitted.
        version: Agent version string — the build this invocation is running. Optional today,
            and omitting it emits a ``UserWarning`` once per agent per process: a
            trace with no version cannot be attributed to a build, and the dashboard records it
            against the version ``unknown``. It becomes required in coalex 2.0.
        prompt_version: Prompt Vault version driving this invocation.
        prompt_type: Prompt type (e.g. ``"system"``).
        run_id: One execution of an agent graph, across interrupts and resumes. Carried from an
            enclosing context when omitted. Never derive it from a framework's thread or session id.
        replay_id: The replay run this invocation belongs to. Carried from an enclosing context when
            omitted.
        case_id: The historical case being replayed. Carried from an enclosing context when omitted.
        case_timestamp: When the original case happened. Timezone-aware; a naive datetime raises
            ``ValueError``, because the attribute must carry an offset. Carried from an enclosing
            context when omitted.
    """
    replay_id, case_id, case_ts = _resolve_replay(replay_id, case_id, case_timestamp)

    outer_agent = _coalex_agent_id.get()
    nested_in_other_agent = outer_agent is not None and outer_agent != agent_id

    if request_id is None:
        request_id = _coalex_request_id.get()
    if run_id is None:
        run_id = _coalex_run_id.get()
    parent_agent_id: str | None = None
    if nested_in_other_agent:
        parent_agent_id = outer_agent
    elif outer_agent == agent_id:
        parent_agent_id = _coalex_parent_agent_id.get()
        if version is None:
            version = _coalex_agent_version.get()

    if version is None:
        _warn_missing_version(agent_id)

    token_a = _coalex_agent_id.set(agent_id)
    token_p = _coalex_parent_agent_id.set(parent_agent_id)
    token_r = _coalex_request_id.set(request_id)
    token_u = _coalex_run_id.set(run_id)
    token_v = _coalex_agent_version.set(version)
    token_rp = _coalex_replay_id.set(replay_id)
    token_c = _coalex_case_id.set(case_id)
    token_ct = _coalex_case_timestamp.set(case_ts)
    # The prompt vars are touched only when given, or when another agent's prompt must not carry
    # over, so a value get_prompt() set for this agent isn't clobbered.
    token_pv = (
        _coalex_prompt_version.set(prompt_version) if prompt_version is not None or nested_in_other_agent else None
    )
    token_pt = _coalex_prompt_type.set(prompt_type) if prompt_type is not None or nested_in_other_agent else None

    try:
        tracer = trace.get_tracer("coalex")

        attrs: dict[str, str | int] = {"coalex.agent_id": agent_id}
        if parent_agent_id is not None:
            attrs["coalex.parent_agent_id"] = parent_agent_id
        if request_id is not None:
            attrs["coalex.request_id"] = request_id
        if run_id is not None:
            attrs["coalex.run_id"] = run_id
        if version is not None:
            attrs["coalex.agent_version"] = version
        if prompt_version is not None:
            attrs["coalex.prompt_version"] = prompt_version
        if prompt_type is not None:
            attrs["coalex.prompt_type"] = prompt_type
        if replay_id is not None:
            attrs["coalex.replay_id"] = replay_id
        if case_id is not None:
            attrs["coalex.case_id"] = case_id
        if case_ts is not None:
            attrs["coalex.case_timestamp"] = case_ts
        attrs.update(invocation_conventions(agent_id))

        with tracer.start_as_current_span(
            "coalex.invocation",
            attributes=attrs,
            record_exception=False,
            set_status_on_exception=False,
        ) as span:
            try:
                yield
            except BaseException as exc:
                set_status_for_exception(span, exc)
                raise
            span.set_status(Status(StatusCode.OK))
    finally:
        _restore(_coalex_agent_id, token_a)
        _restore(_coalex_parent_agent_id, token_p)
        _restore(_coalex_request_id, token_r)
        _restore(_coalex_run_id, token_u)
        _restore(_coalex_agent_version, token_v)
        _restore(_coalex_prompt_version, token_pv)
        _restore(_coalex_prompt_type, token_pt)
        _restore(_coalex_replay_id, token_rp)
        _restore(_coalex_case_id, token_c)
        _restore(_coalex_case_timestamp, token_ct)

current_agent_id 

current_agent_id() -> str | None

Return the agent_id of the active coalex_context(), or None outside one.

evaluate() reads it so a mid-run escalation names its agent before the trace is ingested.

Source code in coalex/context.py 
174
175
176
177
178
179
def current_agent_id() -> str | None:
    """Return the agent_id of the active coalex_context(), or None outside one.

    ``evaluate()`` reads it so a mid-run escalation names its agent before the trace is ingested.
    """
    return _coalex_agent_id.get()

current_agent_version 

current_agent_version() -> str | None

Return the agent version of the active coalex_context(), or None when it has none.

Source code in coalex/context.py 
192
193
194
def current_agent_version() -> str | None:
    """Return the agent version of the active coalex_context(), or None when it has none."""
    return _coalex_agent_version.get()

current_parent_agent_id 

current_parent_agent_id() -> str | None

Return the agent whose context the active one opened inside, or None for a top-level agent.

Source code in coalex/context.py 
197
198
199
def current_parent_agent_id() -> str | None:
    """Return the agent whose context the active one opened inside, or None for a top-level agent."""
    return _coalex_parent_agent_id.get()

current_request_id 

current_request_id() -> str | None

Return the request_id of the active coalex_context(), given or carried from an outer one.

Source code in coalex/context.py 
182
183
184
def current_request_id() -> str | None:
    """Return the request_id of the active coalex_context(), given or carried from an outer one."""
    return _coalex_request_id.get()

current_run_id 

current_run_id() -> str | None

Return the run_id of the active coalex_context(), given or carried from an outer one.

Source code in coalex/context.py 
187
188
189
def current_run_id() -> str | None:
    """Return the run_id of the active coalex_context(), given or carried from an outer one."""
    return _coalex_run_id.get()

from_escalation 

from_escalation(
    escalation: Mapping[str, Any] | Any,
    *,
    resolver: ResolverPolicy | str | None = None,
    resume_hook: ResumeHook | None = None,
    question: str | None = None,
    composer: QuestionComposer | None = None,
) -> DecisionRequest

Materialize an escalation as a Decision Request (spec §4).

Accepts the escalation as the API JSON dict or any object with matching attributes (e.g. the transformer's EscalationItem). The task state and elicitation action follow spec §5's mapping. The escalation has no resolver/resume-hook of its own, so the producer supplies them here; resolver defaults to self.

Parameters:

Name Type Description Default
escalation Mapping[str, Any] | Any

The escalation dict/object (status + fields).

 required
resolver ResolverPolicy | str | None

Resolver policy or its self | role:<name> string. Default self.

 None
resume_hook ResumeHook | None

How the producer resumes after resolution (spec §4.4).

 None
composer QuestionComposer | None

Decides what the card shows (spec §6). Default :class:IdentityComposer.

 None
question str | None

Overrides the composed question. Kept for callers that passed one before the composer existed; a composer is the better place to decide it.

 None

Raises:

Type Description
ValueError

If the escalation carries an unrecognized status.
Source code in coalex/decision_request.py 
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
def from_escalation(
    escalation: Mapping[str, Any] | Any,
    *,
    resolver: ResolverPolicy | str | None = None,
    resume_hook: ResumeHook | None = None,
    question: str | None = None,
    composer: QuestionComposer | None = None,
) -> DecisionRequest:
    """Materialize an escalation as a Decision Request (spec §4).

    Accepts the escalation as the API JSON dict or any object with matching attributes
    (e.g. the transformer's ``EscalationItem``). The task state and elicitation action
    follow spec §5's mapping. The escalation has no resolver/resume-hook of its own, so
    the producer supplies them here; ``resolver`` defaults to ``self``.

    Args:
        escalation: The escalation dict/object (status + fields).
        resolver: Resolver policy or its ``self`` | ``role:<name>`` string. Default ``self``.
        resume_hook: How the producer resumes after resolution (spec §4.4).
        composer: Decides what the card shows (spec §6). Default :class:`IdentityComposer`.
        question: Overrides the composed question. Kept for callers that passed one before the
            composer existed; a composer is the better place to decide it.

    Raises:
        ValueError: If the escalation carries an unrecognized status.
    """
    status = str(_get(escalation, "status", "pending"))
    if status not in _STATUS_TO_STATE:
        raise ValueError(f"unknown escalation status {status!r}; expected one of {sorted(_STATUS_TO_STATE)}")
    task_state, action = _STATUS_TO_STATE[status]

    # Everything the approver reads comes from here and nowhere else (spec §6). The card
    # is generic; deciding what it shows is this stage's job, and routing it through one object is
    # what lets a better composer change the card without the card changing.
    composed = (composer or IdentityComposer()).compose(escalation)
    raw_input = _get(escalation, "input", None)

    context = ContextPayload(
        question=question if question is not None else composed.question,
        output=composed.output,
        input=dict(raw_input) if isinstance(raw_input, Mapping) else None,
        summary=composed.summary,
    )
    schema = composed.schema

    if isinstance(resolver, str):
        resolver_policy = ResolverPolicy.parse(resolver)
    elif resolver is None:
        resolver_policy = ResolverPolicy(kind="self")
    else:
        resolver_policy = resolver

    # The producer may pass the hook explicitly; otherwise adopt the one persisted on the
    # escalation at evaluate(), so the Decision Request API returns it verbatim.
    hook = resume_hook
    if hook is None:
        raw_hook = _get(escalation, "resume_hook", None)
        if isinstance(raw_hook, Mapping):
            hook = ResumeHook.from_dict(raw_hook)

    result: ElicitationResult | None = None
    if action is not None:
        raw_corrections = _get(escalation, "corrections", None)
        result = ElicitationResult(
            action=action,
            content=dict(raw_corrections) if isinstance(raw_corrections, Mapping) else None,
            reason=_get(escalation, "resolution_reason", None),
            reviewer=_get(escalation, "resolved_by", None),
            reviewer_name=_get(escalation, "reviewer_name", None),
            reviewer_email=_get(escalation, "reviewer_email", None),
            resolved_at=_iso(_get(escalation, "resolved_at", None)),
        )

    return DecisionRequest(
        id=str(_get(escalation, "escalation_id", "")),
        task_state=task_state,
        context=context,
        question_schema=schema,
        resolver=resolver_policy,
        account_id=_get(escalation, "account_id", None),
        agent_id=_get(escalation, "agent_id", None),
        # The display name under its own key, from a server that serves it; an older one serves
        # only `agent_name`.
        agent_display_name=_get(escalation, "agent_display_name", None) or _get(escalation, "agent_name", None),
        tool_name=_get(escalation, "tool_name", None),
        request_id=_get(escalation, "request_id", None),
        resume_hook=hook,
        result=result,
        expires_at=_iso(_get(escalation, "expires_at", None)),
        created_at=_iso(_get(escalation, "created_at", None)),
        updated_at=_iso(_get(escalation, "updated_at", None)),
    )

resume_body_from_result 

resume_body_from_result(
    result: ElicitationResult,
) -> dict[str, Any]

Map a resolved ElicitationResult → the JSON body an agent's resume endpoint expects.

The exact wire shape the working PharmaConnect resume path reads back (pilot-bluepharma-v2 agent/src/runtime/api.py ResumeRequest + domain/approval.py request_approval):

{"decision", "corrections", "reason", "reviewer_name", "reviewer_email"}

Mapping from ElicitationAction (spec §5):

  • ACCEPT with no content → decision="approved"
  • ACCEPT with content → decision="corrected", corrections=content
  • DECLINE → decision="rejected", reason carried through verbatim (deny-with-instruction)
  • CANCEL → decision="expired"

Pure: no I/O. The resolve-trigger POSTs this to resume_hook.url to resume the run.

Source code in coalex/decision_request.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def resume_body_from_result(result: ElicitationResult) -> dict[str, Any]:
    """Map a resolved ``ElicitationResult`` → the JSON body an agent's resume endpoint expects.

    The exact wire shape the working PharmaConnect resume path reads back (pilot-bluepharma-v2
    ``agent/src/runtime/api.py`` ``ResumeRequest`` + ``domain/approval.py`` ``request_approval``):

    ``{"decision", "corrections", "reason", "reviewer_name", "reviewer_email"}``

    Mapping from ``ElicitationAction`` (spec §5):

    * ``ACCEPT`` with no ``content`` → ``decision="approved"``
    * ``ACCEPT`` with ``content`` → ``decision="corrected"``, ``corrections=content``
    * ``DECLINE`` → ``decision="rejected"``, ``reason`` carried through verbatim (deny-with-instruction)
    * ``CANCEL`` → ``decision="expired"``

    Pure: no I/O. The resolve-trigger POSTs this to ``resume_hook.url`` to resume the run.
    """
    if result.action is ElicitationAction.ACCEPT:
        decision = "corrected" if result.content else "approved"
    elif result.action is ElicitationAction.DECLINE:
        decision = "rejected"
    else:  # CANCEL
        decision = "expired"

    return {
        "decision": decision,
        "corrections": result.content if decision == "corrected" else None,
        "reason": result.reason,
        "reviewer_name": result.reviewer_name,
        "reviewer_email": result.reviewer_email,
    }

task_state_for_status 

task_state_for_status(status: str) -> TaskState

Map an escalation status to its MCP task state (spec §5).

Source code in coalex/decision_request.py 
457
458
459
460
461
462
def task_state_for_status(status: str) -> TaskState:
    """Map an escalation status to its MCP task state (spec §5)."""
    state = TASK_STATE_BY_STATUS.get(status)
    if state is None:
        raise ValueError(f"unknown escalation status {status!r}; expected one of {sorted(TASK_STATE_BY_STATUS)}")
    return state

decide_policy 

decide_policy(
    *,
    governed: bool,
    risk_tier: str | None = None,
    trust_score: float | None = None,
    blocked: bool = False,
) -> PolicyDecision

Pure policy decision: (governed, risk_tier, trust_score) → allow | escalate | block.

Deterministic and side-effect-free. trust_score is the 0-100 dial. Rules:

  • blocked (or risk_tier == "blocked") → block — never runs.
  • not governed → allow — ungoverned tools always pass (Scene 1 / YOLO).
  • governed + risk_tier == "high" → escalate — a high-risk (destructive) tool always needs a human, regardless of trust; no auto-approve on that path.
  • governed + trust_score >= HIGH_TRUST_MIN → allow — earned autonomy (Scene 4).
  • governed, otherwise → escalate — the default governed behaviour (Scene ⅔).
Source code in coalex/gateway.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def decide_policy(
    *,
    governed: bool,
    risk_tier: str | None = None,
    trust_score: float | None = None,
    blocked: bool = False,
) -> PolicyDecision:
    """Pure policy decision: (governed, risk_tier, trust_score) → allow | escalate | block.

    Deterministic and side-effect-free. ``trust_score`` is the 0-100 dial. Rules:

    * ``blocked`` (or ``risk_tier == "blocked"``) → **block** — never runs.
    * not ``governed`` → **allow** — ungoverned tools always pass (Scene 1 / YOLO).
    * governed + ``risk_tier == "high"`` → **escalate** — a high-risk (destructive) tool always needs
      a human, regardless of trust; no auto-approve on that path.
    * governed + ``trust_score >= HIGH_TRUST_MIN`` → **allow** — earned autonomy (Scene 4).
    * governed, otherwise → **escalate** — the default governed behaviour (Scene 2/3).
    """
    if blocked or risk_tier == "blocked":
        return "block"
    if not governed:
        return "allow"
    if risk_tier == "high":
        return "escalate"
    if trust_score is not None and trust_score >= HIGH_TRUST_MIN:
        return "allow"
    return "escalate"

guard_tool_call 

guard_tool_call(
    *,
    tool_name: str,
    arguments: dict[str, Any],
    governed: bool,
    run: Callable[[], Any],
    request_id: str,
    agent_id: str | None = None,
    risk_tier: str | None = None,
    trust_score: float | None = None,
    resolver: str = "self",
    blocked: bool = False,
    resume_hook: ResumeHook | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> GuardResult

Wrap one tool call in the Gateway v0 policy. Same agent, same prompt — the config (the governed flag) decides whether the wire executes or becomes an escalation.

  • allow → runs run() and records an allowed-call audit row; returns its result.
  • escalate → does NOT run the tool; calls the existing SDK evaluate() (creating a pending Decision Request the Console resolves; pause/resume rides resume_hook + the resolve callback, exactly as PharmaConnect does today). If the risk engine instead auto-approves, the tool runs; if it rejects, the tool is blocked.
  • block → does NOT run the tool; records a blocked-call audit row.

run is a zero-arg callable that executes the underlying tool (so allow/auto-approved actually invoke it, while escalate/block never do). It must be synchronous: an async callable would return an un-awaited coroutine, and this raises rather than record a call that never happened.

This fails closed, and that is deliberate. If Coalex is unreachable at the moment a decision is needed — the escalate path calls evaluate(), which raises on a transport error or a non-2xx — the exception propagates and your tool does not run. A governance layer that waves calls through when it cannot reach its own control plane is not governing.

The behaviour was always this; what was missing was anyone saying so. A developer saw a raw httpx.ConnectError and had to infer, from a stack trace, that their governed action had been prevented. It is now logged in those words before the exception leaves this function.

Source code in coalex/gateway.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def guard_tool_call(
    *,
    tool_name: str,
    arguments: dict[str, Any],
    governed: bool,
    run: Callable[[], Any],
    request_id: str,
    agent_id: str | None = None,
    risk_tier: str | None = None,
    trust_score: float | None = None,
    resolver: str = "self",
    blocked: bool = False,
    resume_hook: ResumeHook | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> GuardResult:
    """Wrap one tool call in the Gateway v0 policy. Same agent, same prompt — the config
    (the ``governed`` flag) decides whether the wire executes or becomes an escalation.

    * **allow** → runs ``run()`` and records an allowed-call audit row; returns its result.
    * **escalate** → does NOT run the tool; calls the existing SDK ``evaluate()`` (creating a pending
      Decision Request the Console resolves; pause/resume rides ``resume_hook`` + the resolve callback,
      exactly as PharmaConnect does today). If the risk engine instead auto-approves, the tool runs;
      if it rejects, the tool is blocked.
    * **block** → does NOT run the tool; records a blocked-call audit row.

    ``run`` is a zero-arg callable that executes the underlying tool (so allow/auto-approved actually
    invoke it, while escalate/block never do). It must be **synchronous**: an async callable would
    return an un-awaited coroutine, and this raises rather than record a call that never happened.

    **This fails closed, and that is deliberate.** If Coalex is unreachable at the moment
    a decision is needed — the ``escalate`` path calls ``evaluate()``, which raises on a transport
    error or a non-2xx — the exception propagates and **your tool does not run**. A governance
    layer that waves calls through when it cannot reach its own control plane is not governing.

    The behaviour was always this; what was missing was anyone saying so. A developer saw a raw
    ``httpx.ConnectError`` and had to infer, from a stack trace, that their governed action had been
    prevented. It is now logged in those words before the exception leaves this function.
    """
    import coalex as _sdk

    cfg = _config or _sdk._config
    api_key = _api_key or _sdk._api_key
    if cfg is None:
        raise RuntimeError("coalex.register() must be called before guard_tool_call()")

    decision = decide_policy(governed=governed, risk_tier=risk_tier, trust_score=trust_score, blocked=blocked)
    policy_inputs: dict[str, Any] = {"governed": governed, "risk_tier": risk_tier, "trust_score": trust_score}

    if decision == "block":
        _record_gateway_audit(
            cfg,
            api_key,
            tool_name=tool_name,
            arguments=arguments,
            policy_decision="block",
            outcome="blocked",
            policy_inputs=policy_inputs,
            request_id=request_id,
        )
        return GuardResult(decision="block")

    if decision == "escalate":
        # Fail-closed, said out loud. Whatever goes wrong reaching Coalex here, the tool does not
        # run — and the developer is told that in those words rather than left to infer it from a
        # transport exception.
        try:
            # The arguments ARE the card, flat. This sent
            # ``{"tool": ..., "arguments": {...}}``, so the approver saw one object-typed field —
            # "[object Object]" — and was asked to release a wire without seeing the amount or the
            # recipient. That was already settled twice, for the demo agent and for
            # ``from_task``; this was the third producer and it never got the message.
            #
            # Flat keys give the card real, individually editable fields, which is also what
            # edit-and-accept needs: a reviewer's corrections come back to the agent and are what
            # the tool is then called with. Against the nested shape those corrections were an
            # object containing an object, and the tool name sat beside them as an editable field.
            #
            # The tool name is NOT in the payload, which is where this differs from the demo agent
            # (api.py keeps a flat ``tool`` key). ``from_task`` leaves it out for a reason that
            # applies here too: what a reviewer edits is what the tool runs with, so anything that
            # is not an argument has no business being editable. It travels in ``metadata``, and
            # ``reason`` states it in words for the card.
            card = {str(k): v for k, v in (arguments or {}).items()}
            reason = f"The agent is requesting approval to run {tool_name} before it executes."
            ev: EvaluationDecision = evaluate(
                request_id=request_id,
                input=card,
                output=card,
                agent_id=agent_id,
                resume_hook=resume_hook,
                resolver=resolver,
                tool_name=tool_name,
                metadata={
                    "gateway": True,
                    "tool": tool_name,
                    "reason": reason,
                    "governed": governed,
                    "risk_tier": risk_tier,
                },
                _config=_config,
                _api_key=_api_key,
            )
        except Exception as exc:
            logger.error(
                "Coalex was unreachable while deciding tool %r (%s: %s) — the tool was NOT executed. "
                "This is fail-closed by design: a governed call does not proceed when the decision "
                "cannot be made.",
                tool_name,
                type(exc).__name__,
                exc,
            )
            raise
        if ev.status == "escalated":
            # Paused: the Console resolves the Decision Request; the agent resumes and runs the tool.
            return GuardResult(decision="escalate", escalation_id=ev.escalation_id)
        # There was a `rejected` branch here, and the server has never been able to reach it.
        # `/evaluate` decides between `escalated` and `auto_approved` and nothing else —
        # `_decide` in routes/evaluate.py has exactly those two outcomes, and EvaluateResponse says
        # so in a comment. A human rejecting an ESCALATION is a different thing at a different
        # moment: it lands on the escalation's own status, not on this response.
        #
        # Removed rather than implemented. A server-side refusal is a product decision — what would
        # refuse, on what grounds, and how the agent is told — not a gap to be filled in by a
        # client that was already pretending to handle it.
        #
        # But taking it out exposed what the dead branch had been hiding: everything that was not
        # `escalated` fell through to running the tool. So a status this client does not
        # understand — a value a future server adds, a typo, a refusal implemented later — was
        # permission to proceed. In a governance SDK, on the branch whose whole job is deciding
        # whether a call may happen, the unknown case has to be the refusing one.
        if ev.status != "auto_approved":
            logger.error(
                "Coalex answered %r for tool %r, which this SDK version does not understand — the "
                "tool was NOT executed. This is fail-closed by design; upgrade the `coalex` package "
                "if your platform is newer than this client.",
                ev.status,
                tool_name,
            )
            _record_gateway_audit(
                cfg,
                api_key,
                tool_name=tool_name,
                arguments=arguments,
                policy_decision="block",
                outcome="unknown_decision",
                policy_inputs={**policy_inputs, "status": ev.status},
                request_id=request_id,
            )
            return GuardResult(decision="block")
        # auto_approved by the risk engine → proceed exactly like an allow.
        #
        # But say so in the audit row. Without this the success row below is `policy_decision=allow`
        # with `governed=True`, which the transformer classifies as `earned` — and the rail then tells
        # a reader that the AGENT'S TRUST SCORE won this call its freedom. It won nothing: the risk
        # engine scored this one call low and released it, and the next one may pause.
        policy_inputs = {**policy_inputs, "auto_approved": True}

    try:
        result = run()
    except Exception as exc:
        # A tool that raises AFTER doing its work — a payment that posted and then failed to parse
        # the response — used to leave no row at all, because this call was not inside a try.
        # The customer's reconciliation and our trail then disagree, and ours is the
        # one that is wrong.
        #
        # The exception TYPE, never its message: a message can carry the tool's arguments, and the
        # audit copy is deliberately redacted.
        _record_gateway_audit(
            cfg,
            api_key,
            tool_name=tool_name,
            arguments=arguments,
            policy_decision="allow",
            outcome="error",
            policy_inputs={**policy_inputs, "error_type": type(exc).__name__},
            # The only one of the four audit calls that omitted it — and precisely the
            # case the comment above says this exists for: a payment that WAS made and then failed
            # to read the response. Without the run id it cannot be tied to the session it happened
            # in, so the row the customer's reconciliation needs is the one nothing can find.
            request_id=request_id,
        )
        raise

    if inspect.isawaitable(result):
        # An async tool passed as `run=lambda: send_wire(...)` returns an un-awaited coroutine. That
        # used to be stored as the result and audited as a success, so an async agent showed a clean
        # run of allowed calls in which NOTHING had executed — the only signal a RuntimeWarning most
        # services filter out.
        #
        # Raising is the only honest answer: the tool has not run, and no audit row may claim it did.
        if hasattr(result, "close"):
            result.close()  # do not leave the un-awaited coroutine warning to fire later
        raise TypeError(
            f"guard_tool_call received an async callable for tool {tool_name!r}. It was not awaited, "
            "so the tool did NOT run and nothing was recorded. Wrap the coroutine in a synchronous "
            "runner, or await it yourself and pass a callable that returns the finished value."
        )

    _record_gateway_audit(
        cfg,
        api_key,
        tool_name=tool_name,
        arguments=arguments,
        policy_decision="allow",
        outcome="success",
        policy_inputs=policy_inputs,
        request_id=request_id,
    )
    return GuardResult(decision="allow", result=result)

declare_inventory 

declare_inventory(
    inventory: AgentInventory,
    *,
    config: CoalexConfig,
    api_key: str,
    default_name: str,
) -> bool

POST one declaration with an explicit config and key. Returns whether it landed; never raises.

coalex.declare() does the same with what register() configured. default_name is treated as an explicitly passed service_name: the id when neither the declaration nor the active context names one (deprecated), and never a default service name.

register() runs at application startup, so this must not be able to stop an agent from starting. A governance registration failing is a reason to log loudly, not a reason for a customer's agent to be down — the agent still traces, and the declaration retries on next boot.

Source code in coalex/inventory.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def declare_inventory(
    inventory: AgentInventory,
    *,
    config: CoalexConfig,
    api_key: str,
    default_name: str,
) -> bool:
    """POST one declaration with an explicit config and key. Returns whether it landed; never raises.

    ``coalex.declare()`` does the same with what ``register()`` configured. ``default_name`` is
    treated as an explicitly passed ``service_name``: the id when neither the declaration nor the
    active context names one (deprecated), and never a default service name.

    ``register()`` runs at application startup, so this must not be able to stop an agent from
    starting. A governance registration failing is a reason to log loudly, not a reason for a
    customer's agent to be down — the agent still traces, and the declaration retries on next boot.
    """
    return _post(inventory, config=config, api_key=api_key, service_name=default_name)

get_prompt 

get_prompt(
    *,
    agent: str,
    name: str,
    version: str | int = "production",
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> PromptVersion

Fetch a prompt template from the Prompt Vault.

Parameters:

Name Type Description Default
agent str

Agent ID (e.g., "sales-copilot").

 required
name str

Prompt name (e.g., "system").

 required
version str | int

"production" (default), "draft", "latest", or a version number. Anything else raises ValueError — in particular "staging", which the Prompt Vault has never had, whatever older documentation said.

 'production'
_config CoalexConfig | None

Override config (testing only).

 None
_api_key str | None

Override API key (testing only).

 None
Source code in coalex/prompts.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def get_prompt(
    *,
    agent: str,
    name: str,
    version: str | int = "production",
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> PromptVersion:
    """Fetch a prompt template from the Prompt Vault.

    Args:
        agent: Agent ID (e.g., "sales-copilot").
        name: Prompt name (e.g., "system").
        version: "production" (default), "draft", "latest", or a version number.
            Anything else raises ``ValueError`` — in particular "staging", which the Prompt
            Vault has never had, whatever older documentation said.
        _config: Override config (testing only).
        _api_key: Override API key (testing only).
    """
    import coalex as _sdk

    version = _checked_version(version)
    cfg = _config or _sdk._config
    api_key = _api_key or _sdk._api_key
    if cfg is None:
        raise RuntimeError("coalex.register() must be called before get_prompt()")

    with httpx.Client(timeout=30.0) as client:
        resp = client.get(
            f"{cfg.endpoint}/api/v2/prompts/{path_segment(agent)}/{path_segment(name)}",
            params={"version": version},
            headers=sdk_headers(api_key),
        )
        raise_for_status(resp, "get_prompt()")

    data = resp.json()
    result = PromptVersion(
        id=data["id"],
        agent_id=data["agent_id"],
        # The Prompt Vault keys a prompt by (agent, prompt_type); there is no separate `name`
        # column, and the URL's `{name}` segment IS the prompt_type. So `name` mirrors it.
        name=data["prompt_type"],
        prompt_type=data["prompt_type"],
        version=data["version"],
        content=data["content"],
        status=data["status"],
        metadata=data.get("metadata", {}),
    )

    _record_prompt_version(result)
    return result

flush_ungoverned 

flush_ungoverned(
    cfg: Any = None, api_key: str | None = None
) -> int

Deliver what the spool is holding. Returns how many landed. Never raises.

Called after a SUCCESSFUL platform call as well as from record, because success is the only honest signal that the platform is back. A spool that only drains when the next failure happens would keep the backlog exactly as long as the outage lasts, and then some.

Source code in coalex/ungoverned.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def flush(cfg: Any = None, api_key: str | None = None) -> int:
    """Deliver what the spool is holding. Returns how many landed. Never raises.

    Called after a SUCCESSFUL platform call as well as from ``record``, because success is the only
    honest signal that the platform is back. A spool that only drains when the next failure happens
    would keep the backlog exactly as long as the outage lasts, and then some.
    """
    endpoint, key = _identity(cfg, api_key)
    if not endpoint or not key:
        return 0
    # Another flush in this process is already sending: it will take what is on disk.
    if not _flush_lock.acquire(blocking=False):
        return 0
    try:
        return _flush_locked(cfg, api_key)
    finally:
        _flush_lock.release()

record_ungoverned 

record_ungoverned(
    *,
    request_id: str,
    action: str,
    error: str,
    _config: Any = None,
    _api_key: str | None = None,
) -> None

Record that action ran without governance, because the platform could not be reached.

Call it from the except around evaluate(), beside the decision to proceed. It does not change that decision and cannot fail it.

Delivery is attempted immediately and spooled to disk if it does not land, which is the usual case: the reason this is being called is that the platform was unreachable a moment ago.

Source code in coalex/ungoverned.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
def record(
    *,
    request_id: str,
    action: str,
    error: str,
    _config: Any = None,
    _api_key: str | None = None,
) -> None:
    """Record that ``action`` ran without governance, because the platform could not be reached.

    Call it from the ``except`` around ``evaluate()``, beside the decision to proceed. It does not
    change that decision and cannot fail it.

    Delivery is attempted immediately and spooled to disk if it does not land, which is the usual
    case: the reason this is being called is that the platform was unreachable a moment ago.
    """
    import coalex as _sdk

    cfg = _config or _sdk._config
    api_key = _api_key if _api_key is not None else _sdk._api_key
    entry = {
        "id": uuid.uuid4().hex,
        "request_id": request_id,
        "action": action,
        "error": str(error)[:500],
        "at": time.time(),
    }

    if cfg is None:
        # No registration means no endpoint to send to and no account to attribute it to. Spooling
        # it would be spooling something undeliverable forever.
        logger.warning("coalex: ungoverned action not recorded — register() has not been called")
        return

    try:
        attempt = _post(cfg, api_key, entry)
        if attempt == _DELIVERED:
            flush_in_background(cfg, api_key)  # it landed, so the platform is up: take the backlog with it
            return
        if attempt == _DISCARD:
            return  # `_post` said why, and keeping it would block everything behind it
        _append_spool([entry], _spool_path(cfg, api_key))
    except Exception:  # never break the caller's except block
        logger.warning("coalex: could not record an ungoverned action", exc_info=True)

register 

register(
    *,
    endpoint: str = "https://traces.coalex.ai",
    api_key: str = "dev",
    service_name: str | None = None,
    filter_non_ai_spans: bool = False,
    suppress_export_errors: bool = True,
    genai_semconv: bool | None = None,
    capture_content: bool = True,
    prompt_refs: bool | None = None,
    declare: AgentInventory
    | Sequence[AgentInventory]
    | None = None,
) -> None

Configure the Coalex OTLP exporter.

Call once at application startup. Sets the global TracerProvider with a BatchSpanProcessor exporting OTLP/HTTP to the Coalex proxy.

The SDK always sends through the proxy (via LB) so that auth validation is in the path. Never send directly to the collector.

Parameters:

Name Type Description Default
endpoint str

Coalex proxy HTTP endpoint (default: cloud).

 'https://traces.coalex.ai'
api_key str

API key for proxy authentication (default: "dev" for local).

 'dev'
service_name str | None

Service name for OTLP resource (default: "coalex-sdk-python"). It is not an agent id: a declaration without agent_id outside any coalex_context() falls back to it only when passed explicitly, a fallback deprecated and removed in coalex 2.0.

 None
filter_non_ai_spans bool

If True, only export AI/LLM spans (default: False).

 False
suppress_export_errors bool

If True, suppress export errors gracefully (default: True).

 True
genai_semconv bool | None

If True, additively emit OTel GenAI attributes (gen_ai.input.messages / gen_ai.output.messages / gen_ai.system_instructions) derived from OpenInference's message attributes — the normalised shape the trace viewer and judge want. Defaults to the COALEX_GENAI_SEMCONV env var (1/true/yes/on), or False when unset.

 None
capture_content bool

If False, export spans without their content: the prompts, the responses, input.value/output.value, the gen_ai.* messages, tool arguments, retrieved documents and embedded text are removed from every span and span event before export (see :mod:coalex.content_capture), as are the keys Pydantic AI and Google ADK write their content under. The text of an exception (exception.message and exception.stacktrace on a span event, and the span's status description) is replaced by [removed]; the span keeps its ERROR status, its exception event and exception.type. Model, token counts, timing, status and the coalex.* attributes are kept. Default True: content is exported.

 True
prompt_refs bool | None

If True (the default), the text of a prompt fetched with :func:get_prompt is exported as a reference to its Prompt Vault version (coalex.prompt.ref) instead of in full, and the platform restores it from the vault when it stores the span (see :mod:coalex.prompt_refs): traces are smaller and carry the same governance data. Pass False, or set COALEX_PROMPT_REFS to 0/false/no/off, to export the text as before. Defaults to the COALEX_PROMPT_REFS env var, or True when unset. Ignored when capture_content is False, which exports no prompt text at all.

 None
declare AgentInventory | Sequence[AgentInventory] | None

Optionally tell the platform where this agent lives and what tools it has, so it appears in the Console without anyone registering it by hand. One :class:AgentInventory or a list of them, one per agent of a multi-agent system; each is sent on its own, so one refusal does not stop the others. Best-effort: a failure is logged and startup continues. Omit it and register() makes no network call at all, exactly as before. :func:declare does the same after register().

 None

Raises:

Type Description
RuntimeError

The options cannot take effect. Either register() was called before with different span options (the exporter chain is built once; the message names each option that differs), or the application's own tracer provider cannot take span processors (the message names each option that would be ignored).

A tracer provider the application installed first (its own OpenTelemetry setup, Logfire, opentelemetry-instrument) is kept: the SDK adds its attribute processor and its exporter to it, so capture_content and the other options apply to what Coalex receives. The application's other exporters on that provider still see every span unmodified, and the provider keeps its own resource, so service_name does not rename it.

Source code in coalex/__init__.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def register(
    *,
    endpoint: str = "https://traces.coalex.ai",
    api_key: str = "dev",
    service_name: str | None = None,
    filter_non_ai_spans: bool = False,
    suppress_export_errors: bool = True,
    genai_semconv: bool | None = None,
    capture_content: bool = True,
    prompt_refs: bool | None = None,
    declare: AgentInventory | Sequence[AgentInventory] | None = None,
) -> None:
    """Configure the Coalex OTLP exporter.

    Call once at application startup. Sets the global TracerProvider
    with a BatchSpanProcessor exporting OTLP/HTTP to the Coalex proxy.

    The SDK always sends through the proxy (via LB) so that auth
    validation is in the path. Never send directly to the collector.

    Args:
        endpoint: Coalex proxy HTTP endpoint (default: cloud).
        api_key: API key for proxy authentication (default: "dev" for local).
        service_name: Service name for OTLP resource (default: "coalex-sdk-python"). It is not an
            agent id: a declaration without ``agent_id`` outside any ``coalex_context()`` falls
            back to it only when passed explicitly, a fallback deprecated and removed in coalex 2.0.
        filter_non_ai_spans: If True, only export AI/LLM spans (default: False).
        suppress_export_errors: If True, suppress export errors gracefully (default: True).
        genai_semconv: If True, additively emit OTel GenAI attributes
            (``gen_ai.input.messages`` / ``gen_ai.output.messages`` / ``gen_ai.system_instructions``)
            derived from OpenInference's message attributes — the normalised shape the trace
            viewer and judge want. Defaults to the ``COALEX_GENAI_SEMCONV`` env var
            (``1``/``true``/``yes``/``on``), or False when unset.
        capture_content: If False, export spans without their content: the prompts, the
            responses, ``input.value``/``output.value``, the ``gen_ai.*`` messages, tool arguments,
            retrieved documents and embedded text are removed from every span and span event
            before export (see :mod:`coalex.content_capture`), as are the keys Pydantic AI and
            Google ADK write their content under. The text of an exception (``exception.message``
            and ``exception.stacktrace`` on a span event, and the span's status description) is
            replaced by ``[removed]``; the span keeps its ERROR status, its ``exception`` event and
            ``exception.type``. Model, token counts, timing, status and the ``coalex.*``
            attributes are kept. Default True: content is exported.
        prompt_refs: If True (the default), the text of a prompt fetched with :func:`get_prompt` is
            exported as a reference to its Prompt Vault version (``coalex.prompt.ref``) instead of
            in full, and the platform restores it from the vault when it stores the span (see
            :mod:`coalex.prompt_refs`): traces are smaller and carry the same governance data. Pass
            False, or set ``COALEX_PROMPT_REFS`` to ``0``/``false``/``no``/``off``, to export the
            text as before. Defaults to the ``COALEX_PROMPT_REFS`` env var, or True when unset.
            Ignored when ``capture_content`` is False, which exports no prompt text at all.
        declare: Optionally tell the platform where this agent lives and what tools it has,
            so it appears in the Console without anyone registering it by hand.
            One :class:`AgentInventory` or a list of them, one per agent of a multi-agent system;
            each is sent on its own, so one refusal does not stop the others. Best-effort: a
            failure is logged and startup continues. Omit it and ``register()`` makes no network
            call at all, exactly as before. :func:`declare` does the same after ``register()``.

    Raises:
        RuntimeError: The options cannot take effect. Either ``register()`` was called before with
            different span options (the exporter chain is built once; the message names each option
            that differs), or the application's own tracer provider cannot take span processors
            (the message names each option that would be ignored).

    A tracer provider the application installed first (its own OpenTelemetry setup, Logfire,
    ``opentelemetry-instrument``) is kept: the SDK adds its attribute processor and its exporter to
    it, so ``capture_content`` and the other options apply to what Coalex receives. The
    application's other exporters on that provider still see every span unmodified, and the
    provider keeps its own resource, so ``service_name`` does not rename it.
    """
    global _config, _api_key, _explicit_service_name, _registered_provider, _registered_options

    import os

    from opentelemetry import trace
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
        OTLPSpanExporter,
    )
    from opentelemetry.sdk.resources import Resource
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter

    from coalex.content_capture import ContentStrippingExporter
    from coalex.filtering_exporter import FilteringSpanExporter
    from coalex.genai_semconv import GenAISemconvExporter
    from coalex.prompt_refs import PromptReferenceExporter
    from coalex.safe_exporter import SafeOTLPSpanExporter
    from coalex.span_processor import CoalexAttributePropagator

    if genai_semconv is None:
        genai_semconv = os.environ.get("COALEX_GENAI_SEMCONV", "").strip().lower() in ("1", "true", "yes", "on")
    if prompt_refs is None:
        prompt_refs = os.environ.get("COALEX_PROMPT_REFS", "").strip().lower() not in ("0", "false", "no", "off")

    explicit_service_name = service_name
    if service_name is None:
        service_name = _DEFAULT_SERVICE_NAME

    options: dict[str, Any] = {
        "endpoint": endpoint,
        "api_key": api_key,
        "service_name": service_name,
        "filter_non_ai_spans": filter_non_ai_spans,
        "suppress_export_errors": suppress_export_errors,
        "genai_semconv": genai_semconv,
        "capture_content": capture_content,
        "prompt_refs": prompt_refs,
    }

    # Decide against the provider that is already installed BEFORE anything is stored: a refused
    # call must leave the first registration exactly as it was, so evaluate() and the spans keep
    # agreeing on one endpoint and key.
    existing = trace.get_tracer_provider()
    ours_already = _registered_provider is not None and existing is _registered_provider
    foreign = not ours_already and not isinstance(existing, trace.ProxyTracerProvider | trace.NoOpTracerProvider)
    if ours_already:
        registered = _registered_options or {}
        changed = [name for name, value in options.items() if registered.get(name) != value]
        if changed:
            raise RuntimeError(
                "coalex.register() was already called with different options: "
                + ", ".join(changed)
                + ". The span exporter is built once, so these cannot take effect on spans. "
                "Call register() once, or call it again with the same options."
            )
    elif foreign and not hasattr(existing, "add_span_processor"):
        raise RuntimeError(
            f"A tracer provider is already installed ({type(existing).__name__}) and it cannot take "
            "span processors, so coalex.register() cannot send spans through it. These options "
            "would be ignored: " + ", ".join(options) + ". Install the Coalex "
            "span processor in that provider yourself, or call register() before it is set up."
        )

    _explicit_service_name = explicit_service_name
    _config = CoalexConfig(endpoint=endpoint, service_name=service_name)
    _api_key = api_key

    resource = Resource({"service.name": service_name})

    base_exporter = OTLPSpanExporter(
        endpoint=f"{_config.endpoint}/v1/traces",
        headers=sdk_headers(api_key),
    )

    # Wrap exporter chain (innermost first): OTLP -> content strip | prompt references -> Safe ->
    # Filtering -> GenAI bridge. The GenAI bridge sits outermost so it enriches spans with gen_ai.*
    # before filtering/OTLP. The strip sits innermost, so the filter decides on the span as
    # instrumented and the gen_ai.* copies the bridge derives are stripped with the rest. Prompt
    # references take its place when content is captured, after the bridge, so its copies are
    # referenced too.
    if not capture_content:
        wire_exporter: SpanExporter = ContentStrippingExporter(base_exporter)
    elif prompt_refs:
        wire_exporter = PromptReferenceExporter(base_exporter)
    else:
        wire_exporter = base_exporter
    safe_exporter = SafeOTLPSpanExporter(wire_exporter, suppress_errors=suppress_export_errors)
    final_exporter: SpanExporter = FilteringSpanExporter(safe_exporter) if filter_non_ai_spans else safe_exporter
    if genai_semconv:
        final_exporter = GenAISemconvExporter(final_exporter)

    if not ours_already:
        provider: Any = existing if foreign else TracerProvider(resource=resource)
        provider.add_span_processor(CoalexAttributePropagator())
        provider.add_span_processor(BatchSpanProcessor(final_exporter))
        if not foreign:
            trace.set_tracer_provider(provider)
        _registered_provider = provider
        _registered_options = options

    if declare is not None:
        from coalex.inventory import declare_many

        # Deliberately last: tracing is configured before we talk to anyone, so an agent whose
        # control plane is unreachable still exports spans.
        declare_many(declare, config=_config, api_key=api_key, service_name=_explicit_service_name)

declare 

declare(inventory: AgentInventory) -> bool

declare(inventory: Sequence[AgentInventory]) -> list[bool]

declare(
    inventory: AgentInventory | Sequence[AgentInventory],
) -> bool | list[bool]

Declare one agent, or several, with the endpoint and key register() configured.

For an agent that declares itself after startup, or declares its children once it knows them. Each declaration is checked locally and sent on its own, so one refusal does not stop the others; each refusal is logged with its reason. Never raises for a refused or failed declaration.

Returns:

Type Description
bool | list[bool]

Whether the declaration landed, or a list of that, in order, when given a list.

Raises:

Type Description
RuntimeError

register() has not been called.
Source code in coalex/__init__.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def declare(inventory: AgentInventory | Sequence[AgentInventory], /) -> bool | list[bool]:
    """Declare one agent, or several, with the endpoint and key ``register()`` configured.

    For an agent that declares itself after startup, or declares its children once it knows them.
    Each declaration is checked locally and sent on its own, so one refusal does not stop the
    others; each refusal is logged with its reason. Never raises for a refused or failed
    declaration.

    Returns:
        Whether the declaration landed, or a list of that, in order, when given a list.

    Raises:
        RuntimeError: ``register()`` has not been called.
    """
    if _config is None or _api_key is None:
        raise RuntimeError("coalex.register() must be called before declare()")
    from coalex.inventory import declare_many

    landed = declare_many(inventory, config=_config, api_key=_api_key, service_name=_explicit_service_name)
    return landed[0] if isinstance(inventory, AgentInventory) else landed