Observability · Guide

Python

What the PyPI package actually ships: a standard-library-only core, one extra per framework, lazy imports that fail with a sentence rather than a traceback, and a decorator that handles all four function shapes. The wiring for each framework has its own guide.

pip install cognipeer-observability

Package	cognipeer-observability
Python	3.9 through 3.13
Dependencies	None, except typing_extensions on Python below 3.11
Typing	Ships py.typed; the public surface is annotated
Licence	MIT

The core ships on the standard library alone on purpose: an observability package must never force a dependency resolution on the application it is watching.

Install and extras

Extras pull in only what you use. Install the one matching the framework you already run.

Command	Pulls in	Guide
pip install cognipeer-observability	nothing	Manual instrumentation
pip install "cognipeer-observability[langchain]"	langchain-core>=0.1.0	LangChain
pip install "cognipeer-observability[langgraph]"	langgraph>=0.1.0, langchain-core	LangGraph
pip install "cognipeer-observability[openai-agents]"	openai-agents>=0.0.1	OpenAI Agents SDK
pip install "cognipeer-observability[claude-agent-sdk]"	claude-agent-sdk>=0.1.0	Claude Agent SDK
pip install "cognipeer-observability[otel]"	opentelemetry-sdk>=1.20.0	OpenTelemetry
pip install "cognipeer-observability[all]"	every integration above	—

There is also a dev extra (pytest, pytest-asyncio, mypy, ruff) for working on the package itself — see Contributing.

Modules

Module	What it holds	Guide
cognipeer_observability	init, get_client, observe, trace, TraceSession, flush, shutdown	API reference
cognipeer_observability.langchain	CognipeerCallbackHandler, cognipeer_config, install_langchain_tracing	LangChain
cognipeer_observability.langgraph	graph_config, trace_graph	LangGraph
cognipeer_observability.openai_agents	CognipeerTracingProcessor, install_openai_agents_tracing	OpenAI Agents SDK
cognipeer_observability.claude_agent_sdk	ClaudeMessageTracer, trace_query	Claude Agent SDK
cognipeer_observability.otel	CognipeerSpanExporter	OpenTelemetry

Every framework import is lazy: importing the core pulls in none of them. Importing a framework module without its dependency installed raises a clear ImportError naming the extra to install —

Install it with `pip install cognipeer-observability[langchain]`.

— rather than a bare traceback from three frames deep inside the framework.

The four forms of @observe

The decorator records one event per call and works on sync functions, async def coroutines, generators and async generators. Nesting is automatic: a decorated function called from inside another becomes its child span, and when no session is active the outermost call opens one and closes it when it returns.

import cognipeer_observability as cognipeer
from cognipeer_observability import observe

cognipeer.init(agent={"name": "research-agent"})

@observe(type="tool_call", tool_name="search")
def search(query: str) -> list[str]:
    return index.query(query)

@observe(type="retrieval")
async def fetch_context(topic: str) -> str:
    ...

@observe(name="stream_answer")
async def stream_answer(prompt: str):
    async for chunk in model.stream(prompt):
        yield chunk

For the two generator forms the event closes when the generator is exhausted, and the yielded chunks are collected into the Output section — so a streaming helper produces one event carrying the whole stream, not one event per chunk.

The event label defaults to the function's __name__ rather than its __qualname__: a qualified name drags in <locals> and the enclosing class, which makes a noisy timeline row for no gain. Pass name= to override. The full option list is in the API reference.

Ambient session context

trace() is a context manager. It opens a session, binds it as the ambient one for the block, and closes it on the way out — marking it error and re-raising untouched if the block throws.

from cognipeer_observability import trace

with trace(name="research-agent", thread_id="conv-42") as session:
    context = fetch_context("quarterly results")
    answer = summarize(context)

Everything inside lands in that session: nested @observe calls, framework handlers, direct session calls. When you need to steer that binding by hand — a worker thread, a callback that arrives outside the block, a session you created yourself — four context helpers are exported:

Helper	Use
use_session(session)	Bind session as the ambient one for the block
use_span(key)	Bind key as the parent span for nested work in the block
get_current_session()	The session enclosing the current execution, or None
get_current_span_key()	The span key enclosing the current execution, or None

The context is carried in contextvars, so it survives await boundaries and keeps concurrent runs apart.

Delivery and exit

Requests run on a background daemon thread (named cognipeer-observability) behind a bounded queue, so recording an event never blocks the traced code and a Console outage cannot back up into your process.

That thread is why a short-lived process has to wait before exiting:

import cognipeer_observability as cognipeer

def handler(event, context):
    with cognipeer.trace(name="lambda-agent"):
        ...
    cognipeer.flush(timeout=10.0)   # or shutdown() to also close open sessions

flush(timeout=10.0) waits for what is queued; shutdown(timeout=10.0) ends every still-open session first, then flushes. Both are safe to call more than once. An atexit hook is registered automatically and covers an ordinary interpreter exit, but a platform that freezes the process — as serverless runtimes do between invocations — can cut it off, so call flush() explicitly there.

Calling init() a second time replaces the process-wide client and flushes the previous one, which is convenient in tests. reset_client() drops it entirely.

Building an integration yourself

TraceSession is the primitive every shipped integration is built on, and it is exported. Where a framework gives you paired start and end callbacks, open_span(key, ...) and close_span(key, ...) turn one pair into exactly one Console event carrying both sides, with the duration measured for you. See the session API for a worked example, and Contributing for the rules — including how to swap the transport so an integration test asserts on the payload instead of hitting the network.

Next

• API reference — every exported symbol and option
• Examples — runnable Python programs, several of them offline
• JavaScript and TypeScript — the other half of the library
• Troubleshooting