peufy-cortex-sdk · Python

Observable cognition infrastructure for AI systems — Python SDK.

Zero required dependencies. Adapters for OpenAI, LangChain, and OpenTelemetry are optional.

Install

pip install peufy-cortex-sdk                  # core only
pip install "peufy-cortex-sdk[openai]"        # + OpenAI adapter
pip install "peufy-cortex-sdk[langchain]"     # + LangChain adapter
pip install "peufy-cortex-sdk[opentelemetry]" # + OpenTelemetry GenAI bridge
pip install "peufy-cortex-sdk[all]"           # everything

Quick start

from cortex import Cortex

cortex = Cortex(
    project_key = "cortex_sk_...",
    endpoint    = "https://cortex.peufy.com",   # or http://localhost:3000
)

session = cortex.session(channel="api")
turn    = session.turn()

turn.input(text="What's the weather in Mumbai?", modality="text")

inf = turn.inference(model="gpt-4o-mini", provider="openai")
try:
    # ... call your LLM ...
    response = openai_client.chat.completions.create(...)
    inf.complete(
        latency_ms = 430,
        tokens     = {"input": 312, "output": 64},
    )
except Exception as e:
    inf.fail(e)
    raise

turn.output(text="It is 34°C and humid.", modality="text")
turn.end()
session.end()

cortex.shutdown()   # flush before process exit

Tool calls

turn.tool(
    name       = "weather_api",
    arguments  = {"city": "Mumbai"},
    result     = {"temp_c": 34, "condition": "humid"},
    status     = "success",
    latency_ms = 180,
)

If status="failure", a linked error.raised event is emitted automatically.

Explicit errors

turn.error(message="Timeout waiting for DB", severity="warning", source="system")

Context managers

with cortex.session(channel="api") as session:
    with session.turn() as turn:
        turn.input(text="Hello", modality="text")
        # ... turn.end() called automatically on __exit__

OpenAI auto-instrumentation

import openai
from cortex.adapters.openai import instrument_openai

client         = openai.OpenAI(api_key="...")
tracked_client = instrument_openai(client, cortex, channel="api")

# Use tracked_client exactly like openai.OpenAI — everything is automatic.
response = tracked_client.chat.completions.create(
    model    = "gpt-4o-mini",
    messages = [{"role": "user", "content": "Hello"}],
)

LangChain auto-instrumentation

from cortex.adapters.langchain import CortexCallbackHandler

handler = CortexCallbackHandler(cortex, channel="agent")

result = chain.invoke(
    {"input": "Summarise this document"},
    config={"callbacks": [handler]},
)

OpenTelemetry GenAI bridge

If your stack already emits OTel traces with GenAI instrumentation (OpenAI/ Anthropic auto-instrumentation, OpenLLMetry, OpenInference, or a hand-rolled tracer following the OTel GenAI semantic conventions), attach this SpanProcessor to your TracerProvider for zero-code-change instrumentation:

from opentelemetry.sdk.trace import TracerProvider
from cortex.adapters.otel import CortexOtelSpanProcessor

provider = TracerProvider()
provider.add_span_processor(CortexOtelSpanProcessor(cortex, channel="api"))

Mapping model: one OTel trace → one Cortex Session (closes when the trace's root span ends); one chat / text_completion span → one Cortex Turn; one execute_tool span → turn.tool() on its parent turn. Both the official semconv (span events gen_ai.content.prompt / .completion) and the Traceloop/OpenLLMetry flattened-attribute convention (gen_ai.prompt.{i}.content) are handled. create_agent / invoke_agent spans and non-GenAI spans are not yet mapped — v1 limitation.

Shutdown

Always call cortex.shutdown() before sys.exit() to flush buffered events.

For serverless / Lambda, call cortex.flush() at the end of each handler instead.