coalex_context()

Context manager that creates a coalex.invocation parent span. All spans created inside this block are tagged with the agent metadata, making them queryable by agent_id, request_id, and version in the Coalex dashboard.

---

Signature

PythonTypeScript

@contextlib.contextmanager
def coalex_context(
    *,
    agent_id: str,
    request_id: str | None = None,
    version: str | None = None,
    prompt_version: int | None = None,
    prompt_type: str | None = None,
    run_id: str | None = None,
) -> Generator[None, None, None]

interface CoalexContextOptions {
    agentId: string;
    requestId?: string;
    runId?: string;
    version?: string;
    promptVersion?: number;
    promptType?: string;
}

async function coalexContext<T>(
    options: CoalexContextOptions,
    fn: () => Promise<T> | T,
): Promise<T>

Callback pattern

The TypeScript SDK uses a callback pattern instead of Python's with statement. Pass your logic as an async function to coalexContext(). A synchronous variant coalexContextSync() is also available.

---

Parameters

Parameter	Type	Default	Description
agent_id	str	required	Identifier for the AI agent. Used to group all traces from the same agent in the dashboard.
request_id	str \| None	None	Your correlation key for one unit of business work. It may span traces, resumes and agents. Carried from an enclosing context when omitted.
version	str \| None	None	Agent version string (e.g., "1.2.0", "canary"): the build this invocation is running. Omitting it warns once per agent; it becomes required in coalex 2.0 / @coalex-ai/sdk 3.0.
prompt_version	int \| None	None	Prompt Vault version driving this invocation.
prompt_type	str \| None	None	Prompt type (e.g., "system").
run_id	str \| None	None	One execution of an agent graph, across interrupts and resumes. Carried from an enclosing context when omitted. Never derive it from a framework's thread, session or conversation id.

All parameters are keyword-only (enforced by *).

---

Returns

A context manager that yields None. Use it with with statements.

---

Span Attributes

The coalex.invocation parent span carries these attributes:

Attribute	Set When	Value
coalex.agent_id	Always	Value of agent_id parameter
coalex.parent_agent_id	Opened inside a context with a different agent	The enclosing agent's agent_id
coalex.request_id	Given, or carried from an enclosing context	The request id
coalex.run_id	Given, or carried from an enclosing context	The run id
coalex.agent_version	Given, or inherited from the same agent	Value of version parameter
coalex.prompt_version / coalex.prompt_type	Given, set by get_prompt(), or inherited from the same agent	The prompt version and type

It also carries the OpenInference and OpenTelemetry GenAI conventions for an agent run, so a third-party backend (Phoenix, Langfuse, any GenAI-convention reader) shows it as an agent:

Attribute	Value
openinference.span.kind	AGENT
gen_ai.operation.name	invoke_agent
gen_ai.agent.name	Value of agent_id parameter

These three are on coalex.invocation only and are not propagated: a descendant is a model call, a tool or a chain. gen_ai.agent.id is not set, because the OpenTelemetry convention reserves it for hosted agents.

Attribute propagation

The CoalexAttributePropagator (configured by register()) puts the coalex.* attributes on every span started while the context is active: every LLM call, retrieval and tool invocation inside the block carries the agent metadata without manual tagging. The values come from the active context first, and from the parent span only outside any context. A framework instrumentor such as OpenInference LangChain parents its spans on its own run tree, so a span's parent can belong to an outer agent while an inner context is active; the context decides. A value a span already carries is never overwritten.

---

How It Works

1. Obtains a tracer named "coalex" from the global TracerProvider.
2. Starts a new span named "coalex.invocation" as the current span.
3. Sets the coalex.* attributes on that span.
4. All spans created inside the with block become children of the coalex.invocation span.
5. When the block exits, the span ends automatically: OK on success, ERROR with an exception event on an exception, and OK on a human-approval pause.

graph TD
    A["coalex.invocation<br/>(coalex.agent_id=support-bot)"] --> B["openai.chat.completions<br/>(auto-instrumented)"]
    A --> C["knowledge_base<br/>(@retrieval_span)"]
    C --> D["embed<br/>(@embedding_span)"]

---

Examples

Basic usage

PythonTypeScript

import coalex
from openai import OpenAI

coalex.register(endpoint="https://traces.coalex.ai", api_key="your-key")
coalex.auto_instrument()

client = OpenAI()

with coalex.coalex_context(agent_id="support-bot", request_id="req-001"):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "What is the refund policy?"}],
    )
    print(response.choices[0].message.content)

import { register, coalexContext, autoInstrument } from "@coalex-ai/sdk";
import OpenAI from "openai";

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

const client = new OpenAI();

await coalexContext({ agentId: "support-bot", requestId: "req-001" }, async () => {
    const response = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: "What is the refund policy?" }],
    });
    console.log(response.choices[0].message.content);
});

With version tracking

PythonTypeScript

with coalex.coalex_context(
    agent_id="claims-processor",
    request_id="claim-2024-0042",
    version="2.1.0",
):
    # All spans inside here carry:
    # - coalex.agent_id = "claims-processor"
    # - coalex.request_id = "claim-2024-0042"
    # - coalex.agent_version = "2.1.0"
    result = process_claim(claim_data)

await coalexContext({
    agentId: "claims-processor",
    requestId: "claim-2024-0042",
    version: "2.1.0",
}, async () => {
    // All spans inside here carry:
    // - coalex.agent_id = "claims-processor"
    // - coalex.request_id = "claim-2024-0042"
    // - coalex.agent_version = "2.1.0"
    const result = await processClaim(claimData);
});

Nested contexts

You can nest contexts when one agent delegates to another in the same process. The inner context creates its own coalex.invocation span as a child of the outer one, and every span inside it belongs to the inner agent, in both SDKs:

• The inner context wins for every value it is given.
• A different agent gets coalex.parent_agent_id set to the enclosing agent, on its coalex.invocation and every descendant. It carries request_id and run_id from the enclosing context unless given, and inherits neither version nor the prompt: those belong to the enclosing agent, so pass the worker's own.
• The same agent nested inherits everything and sets no coalex.parent_agent_id.
• When the inner context exits, the enclosing one is active again.

PythonTypeScript

with coalex.coalex_context(agent_id="orchestrator", request_id="req-100", run_id="run-7", version="2.0.0"):
    # coalex.invocation: agent_id=orchestrator

    with coalex.coalex_context(agent_id="sub-agent-a", version="1.3.0"):
        # coalex.invocation: agent_id=sub-agent-a, parent_agent_id=orchestrator,
        # request_id=req-100, run_id=run-7, agent_version=1.3.0
        run_sub_agent_a()

    with coalex.coalex_context(agent_id="sub-agent-b", version="0.9.1"):
        run_sub_agent_b()

await coalexContext({ agentId: "orchestrator", requestId: "req-100", runId: "run-7", version: "2.0.0" }, async () => {
    // coalex.invocation: agent_id=orchestrator

    await coalexContext({ agentId: "sub-agent-a", version: "1.3.0" }, async () => {
        // coalex.invocation: agent_id=sub-agent-a, parent_agent_id=orchestrator,
        // request_id=req-100, run_id=run-7, agent_version=1.3.0
        await runSubAgentA();
    });

    await coalexContext({ agentId: "sub-agent-b", version: "0.9.1" }, async () => {
        await runSubAgentB();
    });
});

The TypeScript context lives in one AsyncLocalStorage store, so concurrent invocations of different agents in one process each keep their own agent, version and prompt. coalex.parent_agent_id is what the agent claims about its parent: it is displayed, and only a parent the organisation confirms governs.

One agent per LangGraph node

agent_node / agentNode wraps a LangGraph node function so everything the node does belongs to its own agent. Wrap the node, not the graph's invoke: LangGraph re-executes the node from the top on every resume, so the context opens again for the resumed work, and the node's checkpoint namespace is unchanged. The wrapper keeps the function's signature (LangGraph inspects it to pass config) and passes every argument through.

PythonTypeScript

graph.add_node("w1_rfq", coalex.agent_node("rfq-worker", rfq_worker, version="1.4.0"))
graph.add_node("w2_quote", coalex.agent_node("quote-worker", quote_worker, version="1.4.0"))
app = graph.compile(checkpointer=checkpointer)

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

Sync and async node functions are both supported.

graph.addNode("w1_rfq", agentNode("rfq-worker", rfqWorker, { version: "1.4.0" }));
graph.addNode("w2_quote", agentNode("quote-worker", quoteWorker, { version: "1.4.0" }));
const app = graph.compile({ checkpointer });

await coalexContext({ agentId: "orchestrator", requestId, runId, version: "2.0.0" }, () =>
    app.invoke(state, config),
);

A pause is not an error

LangGraph pauses for a human (interrupt()) and hands off to a parent graph (Command(graph=Command.PARENT)) by raising an exception through the node. When the exception passing through a coalex.invocation is a GraphBubbleUp, GraphInterrupt or ParentCommand (or a subclass, such as NodeInterrupt), the span ends OK with no exception event, and the exception is re-raised unchanged so LangGraph still pauses. Every other exception still ends the span ERROR. The spans of the tool, guardrail and retrieval decorators (tool_span, guardrail_span, retrieval_span in Python; toolSpan, guardrailSpan, retrievalSpan in TypeScript) follow the same rule, so a run waiting for an approval inside a decorated tool shows an OK tool span, not a failed one; a paused guardrail records no guardrail.result. The Python embedding and reranker spans do the same. The SDKs recognise the three by class name (Python: anywhere in the MRO; TypeScript: the error's name or any class in its prototype chain), so neither depends on LangGraph.

Reading the active context

Both SDKs expose the active context's values, for example to put them in your own logs or to pass them to a call that runs outside the context. Each returns None / undefined outside a context.

Python	TypeScript	Returns
coalex.current_agent_id()	currentAgentId()	The active agent
coalex.current_parent_agent_id()	currentParentAgentId()	The enclosing agent, for a context nested in a different agent
coalex.current_request_id()	currentRequestId()	The request id, given or carried
coalex.current_run_id()	currentRunId()	The run id, given or carried
coalex.current_agent_version()	currentAgentVersion()	The active agent's version

Generating request IDs

PythonTypeScript

Use uuid4 to generate unique request IDs:

import uuid

with coalex.coalex_context(
    agent_id="my-agent",
    request_id=str(uuid.uuid4()),
):
    ...

Use crypto.randomUUID() to generate unique request IDs:

import { randomUUID } from "node:crypto";

await coalexContext({
    agentId: "my-agent",
    requestId: randomUUID(),
}, async () => {
    // ...
});

---

Notes

• coalex_context() requires register() to have been called first (so a global TracerProvider exists). If register() has not been called, spans are created on a no-op tracer and silently discarded.
• The context manager is not reentrant -- do not reuse a single coalex_context instance. Create a new one for each invocation.
• The request_id does not need to be globally unique, but it should be unique within a reasonable time window (e.g., per user session or per API request).

---

API Reference

coalex.context.coalex_context

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

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

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

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

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

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

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

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

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

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

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

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

Parameters:

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

Source code in coalex/context.py

@contextlib.contextmanager
def coalex_context(
    *,
    agent_id: str,
    request_id: str | None = None,
    version: str | None = None,
    prompt_version: int | None = None,
    prompt_type: str | None = None,
    run_id: str | None = None,
    replay_id: str | None = None,
    case_id: str | None = None,
    case_timestamp: datetime | None = None,
) -> Generator[None, None, None]:
    """Create a parent span with Coalex attributes for all child spans.

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

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

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

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

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

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

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

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

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

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

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

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

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

    if version is None:
        _warn_missing_version(agent_id)

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

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

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

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

coalex.context.agent_node

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

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

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

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

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

Source code in coalex/context.py

def agent_node(agent_id: str, fn: _F, *, version: str | None = None) -> _F:
    """Wrap a LangGraph node function so everything it does belongs to ``agent_id``.

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

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

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

    if inspect.iscoroutinefunction(fn):

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

        return cast("_F", async_node)

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

    return cast("_F", node)