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://your-org.coalex.ai",
    api_key="your-api-key",
    service_name="my-agent",
)

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

register({
    endpoint: "https://your-org.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", "semantic_similarity"]},
)

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", "semantic_similarity"] },
});

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", "semantic_similarity"]},
    )

    # 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", "semantic_similarity"] },
    });

    // 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 
12
13
14
15
16
17
18
@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 
41
42
43
44
45
46
47
48
49
50
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 
53
54
55
56
57
58
59
60
61
@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)

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
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")
        # 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]:
        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)
            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)
            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)
            return dict(self._keys)

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

    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:
                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 
109
110
111
112
113
114
115
116
117
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()

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 
67
68
69
70
71
72
73
74
75
76
77
78
79
@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 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@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.

Source code in coalex/decision_request.py 
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
369
370
371
372
373
374
375
376
377
378
379
380
381
@dataclasses.dataclass(frozen=True)
class DecisionRequest:
    """The gateway-proof HITL seam (spec §4). See module docstring."""

    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 (COA-1999) 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
    # COA-1287: 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

    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_name is not None:
            out["agent_name"] = self.agent_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 
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
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_name is not None:
        out["agent_name"] = self.agent_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 
114
115
116
117
118
119
120
121
122
123
124
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 
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
@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 filtered them at render since COA-1739 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 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
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 filtered them at render since COA-1739 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 {}
        output = {k: v for k, v in dict(raw_output).items() if k.lower() not in FORBIDDEN_FIELD_KEYS}
        raw_metrics = _get(escalation, "metrics", None)
        return ComposedQuestion(
            question=DEFAULT_QUESTION,
            # A human-readable reason the producer curated for the card (COA-1758), 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 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
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 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@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 
170
171
172
173
174
175
176
177
178
179
180
@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 
182
183
184
185
186
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 
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
@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``, COA-1761) 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, COA-1761) 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 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
@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``, COA-1761) 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 / COA-1278).

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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
class TaskState(StrEnum):
    """MCP task states carried by a Decision Request (spec §5 / COA-1278).

    ``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 
36
37
38
39
40
41
42
43
44
45
@dataclasses.dataclass(frozen=True)
class EvaluationDecision:
    """Result of an evaluate() call."""

    status: str  # "auto_approved" | "escalated" | "rejected"
    risk_score: float  # 0.0-1.0
    escalation_id: str | None  # present when status == "escalated"
    # The same outcome in MCP vocabulary (COA-1789): "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 
69
70
71
72
73
74
75
76
77
78
79
80
@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.

Parameters:

Name Type Description Default
endpoint str

Where a turn is POSTed — the URL the Console will talk to.

 required
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

The agent's name on the platform. Defaults to register()'s service_name, which is what its traces are already attributed to — so the declared agent and the observed one are the same row rather than two.

 None
Source code in coalex/inventory.py 
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
@dataclasses.dataclass(frozen=True)
class AgentInventory:
    """What an agent declares about itself.

    Args:
        endpoint: Where a turn is POSTed — the URL the Console will talk to.
        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: The agent's name on the platform. Defaults to ``register()``'s ``service_name``,
            which is what its traces are already attributed to — so the declared agent and the
            observed one are the same row rather than two.
    """

    endpoint: str
    tools: Sequence[str] = ()
    resume_url: str | None = None
    name: str | None = None

    def payload(self, *, default_name: str, sdk: str) -> dict:
        body: dict = {
            "name": self.name or default_name,
            "endpoint": self.endpoint,
            "tools": list(self.tools),
            "sdk": sdk,
        }
        if self.resume_url:
            body["resume_hook"] = {"kind": "checkpoint", "url": self.resume_url}
        return body

PromptVersion dataclass

A single prompt version returned by the Prompt Vault.

Source code in coalex/prompts.py 
12
13
14
15
16
17
18
19
20
21
22
23
@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" | "staging" | "production"
    metadata: dict

ResolutionResult dataclass

Result of a resolve() call.

Source code in coalex/resolve.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@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 (COA-1789). 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 | 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 COA-264'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 | 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 
21
22
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
def declare_agent(
    *,
    agent_id: str,
    display_name: str | None = None,
    metadata: dict | 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 COA-264'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 = {"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={"Authorization": f"Bearer {api_key}"},
        )
        resp.raise_for_status()

    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 
259
260
261
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 
254
255
256
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 
244
245
246
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 
239
240
241
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 
234
235
236
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 
229
230
231
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 
249
250
251
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,
    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, so one org's Console cannot reach another org's agent with a token it was legitimately given.

 required
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 
 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
def verify_console_call(
    token: str | None,
    *,
    audience: str,
    public_keys: Mapping[str, str] | str,
    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, so one org's
            Console cannot reach another org's agent with a token it was legitimately given.
        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:  # noqa: BLE001 - 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}")

    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")

    account_id = claims.get("sub")
    if not isinstance(account_id, str) or not account_id:
        raise ConsoleAuthError("no subject")

    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,
    )

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,
) -> 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.

Context vars are set before the span is created so that CoalexAttributePropagator can propagate attributes even across trace boundaries (e.g. when LangChain auto-instrumentors create new root spans with separate trace IDs).

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 (COA-1980). 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 (COA-1339). When omitted, any value already set by get_prompt() is left untouched.

Parameters:

Name Type Description Default
agent_id str

Identifier for the AI agent.

 required
request_id str | None

Unique request/invocation ID.

 None
version str | None

Agent version string.

 None
prompt_version int | None

Prompt Vault version driving this invocation.

 None
prompt_type str | None

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

 None
Source code in coalex/context.py 
 57
 58
 59
 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
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
@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,
) -> 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.

    Context vars are set before the span is created so that
    ``CoalexAttributePropagator`` can propagate attributes even
    across trace boundaries (e.g. when LangChain auto-instrumentors
    create new root spans with separate trace IDs).

    **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 (COA-1980). 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
    (COA-1339). When omitted, any value already set by ``get_prompt()`` is left
    untouched.

    Args:
        agent_id: Identifier for the AI agent.
        request_id: Unique request/invocation ID.
        version: Agent version string.
        prompt_version: Prompt Vault version driving this invocation.
        prompt_type: Prompt type (e.g. ``"system"``).
    """
    token_a = _coalex_agent_id.set(agent_id)
    token_r = _coalex_request_id.set(request_id)
    token_v = _coalex_agent_version.set(version)
    # Only touch the prompt context vars when explicitly given, so a value set by
    # get_prompt() isn't clobbered when the caller doesn't pass one.
    token_pv = _coalex_prompt_version.set(prompt_version) if prompt_version is not None else None
    token_pt = _coalex_prompt_type.set(prompt_type) if prompt_type is not None else None

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

        attrs: dict[str, str | int] = {"coalex.agent_id": agent_id}
        if request_id is not None:
            attrs["coalex.request_id"] = request_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

        with tracer.start_as_current_span("coalex.invocation", attributes=attrs) as span:
            yield
            span.set_status(Status(StatusCode.OK))
    finally:
        _restore(_coalex_agent_id, token_a)
        _restore(_coalex_request_id, token_r)
        _restore(_coalex_agent_version, token_v)
        _restore(_coalex_prompt_version, token_pv)
        _restore(_coalex_prompt_type, token_pt)

current_agent_id 

current_agent_id() -> str | None

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

Lets evaluate() propagate the agent_id for mid-run escalations (LangGraph interrupt → evaluate before the trace ingests), which the server cannot yet resolve from silver_traces/silver_requests (COA-1076).

Source code in coalex/context.py 
47
48
49
50
51
52
53
54
def current_agent_id() -> str | None:
    """Return the agent_id of the active coalex_context(), or None outside one.

    Lets evaluate() propagate the agent_id for mid-run escalations (LangGraph
    interrupt → evaluate before the trace ingests), which the server cannot yet
    resolve from silver_traces/silver_requests (COA-1076).
    """
    return _coalex_agent_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, COA-1278).

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 
649
650
651
652
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
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
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
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, COA-1278).

    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, COA-1280). 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() (COA-1761), 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),
        agent_name=_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 (COA-1668).

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 (COA-1671) POSTs this to resume_hook.url to resume the run.

Source code in coalex/decision_request.py 
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
def resume_body_from_result(result: ElicitationResult) -> dict[str, Any]:
    """Map a resolved ``ElicitationResult`` → the JSON body an agent's resume endpoint expects (COA-1668).

    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 (COA-1671) 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 
407
408
409
410
411
412
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 (COA-1751): (governed, risk_tier, trust_score) → allow | escalate | block.

Deterministic and side-effect-free. trust_score is the 0-100 dial (COA-1345/COA-1795). 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 
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
def decide_policy(
    *,
    governed: bool,
    risk_tier: str | None = None,
    trust_score: float | None = None,
    blocked: bool = False,
) -> PolicyDecision:
    """Pure policy decision (COA-1751): (governed, risk_tier, trust_score) → allow | escalate | block.

    Deterministic and side-effect-free. ``trust_score`` is the 0-100 dial (COA-1345/COA-1795). 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 (COA-1756). 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 (COA-2181). 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 
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
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 (COA-1756). 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 (COA-2181).** 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 (COA-2181).
        try:
            ev: EvaluationDecision = evaluate(
                request_id=request_id,
                input={"tool": tool_name, "arguments": arguments},
                output={"tool": tool_name, "arguments": arguments},
                agent_id=agent_id,
                resume_hook=resume_hook,
                resolver=resolver,
                metadata={"gateway": True, "tool": tool_name, "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)
        if ev.status == "rejected":
            _record_gateway_audit(
                cfg,
                api_key,
                tool_name=tool_name,
                arguments=arguments,
                policy_decision="block",
                outcome="rejected",
                policy_inputs=policy_inputs,
                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 (COA-2241).
        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
        # (COA-2181). 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 (COA-2241) — 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 (COA-2181).
        #
        # 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 the declaration. Returns whether it landed; never raises.

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 
 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
def declare_inventory(
    inventory: AgentInventory,
    *,
    config: CoalexConfig,
    api_key: str,
    default_name: str,
) -> bool:
    """POST the declaration. Returns whether it landed; never raises.

    ``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.
    """
    try:
        with httpx.Client(timeout=_TIMEOUT_SECONDS) as client:
            resp = client.post(
                f"{config.endpoint}/api/v1/agent-inventory",
                json=inventory.payload(default_name=default_name, sdk=_sdk_version()),
                headers={"Authorization": f"Bearer {api_key}"},
            )
        if resp.status_code >= 400:
            # 422 here means the declaration itself is wrong (a policy field, a bad endpoint).
            # Say so with the server's reason — silence would leave the agent believing it
            # registered, which is the failure mode this whole change exists to remove.
            logger.warning(
                "coalex: agent self-declaration rejected (HTTP %s): %s",
                resp.status_code,
                resp.text[:500],
            )
            return False
    except Exception as exc:  # noqa: BLE001 - startup must survive any transport failure
        logger.warning("coalex: agent self-declaration failed (%s); the agent still traces", exc)
        return False
    return True

get_prompt 

get_prompt(
    *,
    agent: str,
    name: str,
    version: str = "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

"production" (default), "staging", "draft", "latest", or integer.

 '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 
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
71
72
73
def get_prompt(
    *,
    agent: str,
    name: str,
    version: str = "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), "staging", "draft", "latest", or integer.
        _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 get_prompt()")

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

    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 
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
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.
    """
    import coalex as _sdk

    cfg = cfg or _sdk._config
    api_key = api_key if api_key is not None else _sdk._api_key
    if cfg is None:
        return 0
    pending = _read_spool()
    if not pending:
        return 0

    delivered = 0
    remaining: list[dict[str, Any]] = []
    stopped = False
    for i, entry in enumerate(pending):
        # Stop at the first failure: the platform is still down, and sending the rest of a backlog
        # at a service that just refused one helps nobody. The batch cap does the same for a very
        # long outage, so one call never spends a minute draining.
        if stopped or i >= _FLUSH_BATCH:
            remaining.append(entry)
            continue
        if _post(cfg, api_key, {**entry, "recorded_late": True}):
            delivered += 1
        else:
            stopped = True
            remaining.append(entry)

    if delivered:
        _write_spool(remaining)
    return delivered

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 
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
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 = {"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:
        if _post(cfg, api_key, entry):
            flush(cfg, api_key)  # it landed, so the platform is up: take the backlog with it
            return
        _append_spool(entry)
    except Exception:  # noqa: BLE001 — 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.azure.coalex.ai",
    api_key: str = "dev",
    service_name: str = "coalex-sdk-python",
    filter_non_ai_spans: bool = False,
    suppress_export_errors: bool = True,
    genai_semconv: bool | None = None,
    declare: 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.azure.coalex.ai'
api_key str

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

 'dev'
service_name str

Service name for OTLP resource (default: "coalex-sdk-python").

 'coalex-sdk-python'
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 (COA-380). Defaults to the COALEX_GENAI_SEMCONV env var (1/true/yes/on), or False when unset.

 None
declare AgentInventory | None

Optionally tell the platform where this agent lives and what tools it has (COA-1969), so it appears in the Console without anyone registering it by hand. Best-effort: a failure is logged and startup continues. Omit it and register() makes no network call at all, exactly as before.

 None
Source code in coalex/__init__.py 
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
def register(
    *,
    endpoint: str = "https://traces.azure.coalex.ai",
    api_key: str = "dev",
    service_name: str = "coalex-sdk-python",
    filter_non_ai_spans: bool = False,
    suppress_export_errors: bool = True,
    genai_semconv: bool | None = None,
    declare: 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").
        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 (COA-380). Defaults to the ``COALEX_GENAI_SEMCONV`` env var
            (``1``/``true``/``yes``/``on``), or False when unset.
        declare: Optionally tell the platform where this agent lives and what tools it has
            (COA-1969), so it appears in the Console without anyone registering it by hand.
            Best-effort: a failure is logged and startup continues. Omit it and ``register()``
            makes no network call at all, exactly as before.
    """
    global _config, _api_key

    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

    from coalex.filtering_exporter import FilteringSpanExporter
    from coalex.genai_semconv import GenAISemconvExporter
    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")

    _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={"Authorization": f"Bearer {api_key}"},
    )

    # Wrap exporter chain (innermost first): OTLP -> Safe -> Filtering -> GenAI bridge.
    # The GenAI bridge sits outermost so it enriches spans with gen_ai.* before filtering/OTLP.
    safe_exporter = SafeOTLPSpanExporter(base_exporter, suppress_errors=suppress_export_errors)
    final_exporter = FilteringSpanExporter(safe_exporter) if filter_non_ai_spans else safe_exporter
    if genai_semconv:
        final_exporter = GenAISemconvExporter(final_exporter)

    provider = TracerProvider(resource=resource)
    provider.add_span_processor(CoalexAttributePropagator())
    provider.add_span_processor(BatchSpanProcessor(final_exporter))
    trace.set_tracer_provider(provider)

    if declare is not None:
        from coalex.inventory import declare_inventory as _declare

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