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¶

PythonTypeScript

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.

PythonTypeScript

import coalex

coalex.register(
    api_key="your-api-key",
    service_name="my-agent",
)

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

register({
    apiKey: "your-api-key",
    serviceName: "my-agent",
});

Endpoint is optional

The SDK defaults to the Azure-hosted cloud (traces.azure.coalex.ai). Only set endpoint for on-prem, GCP deployments, or local development.

2. Auto-instrument -- patch LLM frameworks¶

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

PythonTypeScript

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.

PythonTypeScript

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 (optional) -- register before traces¶

Pre-register your agent so the dashboard recognizes it before traces arrive. This step is optional but recommended for a better dashboard experience.

PythonTypeScript

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.

PythonTypeScript

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.

PythonTypeScript

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¶

PythonTypeScript

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

# 1. Register
coalex.register(
    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({
    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

@dataclasses.dataclass(frozen=True)
class AgentDeclaration:
    """Result of a declare_agent() call."""

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

EvaluationDecision `dataclass` ¶

Result of an evaluate() call.

Source code in coalex/evaluate.py

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

PromptVersion `dataclass` ¶

A single prompt version returned by the Prompt Vault.

Source code in coalex/prompts.py

@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

@dataclasses.dataclass(frozen=True)
class ResolutionResult:
    """Result of a resolve() call."""

    escalation_id: str
    status: str  # "approved" | "rejected" | "corrected"
    metrics: list[MetricResult]
    resolved_at: str
    resolved_by: str | None = None
    reviewer_name: str | None = None
    reviewer_email: 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

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