komodor-agentops

Python SDK for AgentOps observability, including the worker runtime, with optional extras for framework integrations (LangChain, Claude Agent SDK, ADK, Agno).

Install

pip install komodor-agentops

# With framework adapters
pip install komodor-agentops[langchain]
pip install komodor-agentops[claude-code]
pip install komodor-agentops[adk]
pip install komodor-agentops[agno]
pip install komodor-agentops[all]           # Everything

Quick Start

@observe() decorator

Wrap functions to emit span events automatically:

from komodor_agentops import observe

@observe(name="summarize", as_type="llm")
async def summarize(text: str) -> str:
    ...

AgentOps client (lightweight event buffer)

from komodor_agentops import AgentOps

client = AgentOps(agent_id="my-agent", endpoint="http://localhost:8000")
await client.log("Processing started", run_id="run_1")
await client.flush()

TransportClient (full controlplane client)

For direct controlplane interaction (heartbeat, run lifecycle, event ingest):

from komodor_agentops import TransportClient, AgentSpec

agent = AgentSpec(agent_id="my-agent", agent_card={"name": "My Agent"})
async with TransportClient(endpoint="http://localhost:8000", worker_id="wrk_1", agent=agent) as ops:
    await ops.heartbeat()
    run = await ops.start_run_direct("run_1", input_payload={"prompt": "hello"})
    await ops.emit_span_start(run_id="run_1", span_id="s1", name="tool", span_kind="tool")
    await ops.emit_span_end(run_id="run_1", span_id="s1", name="tool", span_kind="tool")
    await ops.complete_run("run_1", output={"result": "done"})

AgentOpsWorker (unified server)

Run a worker with A2A protocol + AgentOps streaming:

from pathlib import Path

from komodor_agentops import AgentOpsWorker, AgentSpec, A2AMessage, A2ATask

spec = AgentSpec.from_dir(
    Path(__file__).parent,
    agent_card={"skills": [{"id": "search", "name": "search"}]},
)

async def handler(message: A2AMessage) -> A2ATask:
    return A2ATask(id=message.task_id, status={"state": "completed"})

worker = AgentOpsWorker(agent=spec, handler=handler)
await worker.serve(host="0.0.0.0", port=8010)

The standard AgentOps agent layout is:

my_agent/
  agent-spec.yaml
  agent.md
  worker.py
  skills/
    triage.md
    rca/SKILL.md

AgentSpec.from_dir(Path(__file__).parent, ...) requires agent-spec.yaml, then loads agent.md and skills/ automatically. The worker registers loaded skills during heartbeat, and SDK-owned LLM adapters can prepend the agent context at invocation time.

schema_version: 1
agent_id: my-agent
name: My Agent
description: Does useful work.
owner: AgentOps
repo: https://github.com/komodorio/agentops
source_path: packages/workers/my_agent
labels:
  category: example

agent-spec.yaml can also declare triggers — synced to the control plane on heartbeat and shown on the Fleet → Triggers surface. Supported types are schedule (cron), webhook (inbound HTTP endpoint), and slack_channel (subscribe the agent to Slack channels by NAME, not ID — a Slack router worker fetches the aggregated agent → channels map from GET /api/v1/workers/slack-subscriptions and dispatches matching messages):

triggers:
  - id: incidents-sub
    type: slack_channel
    name: Incident channels
    channels:
      - "#incidents" # normalized to "incidents" — lowercase, no "#"
      - alerts

To make a worker available in the AgentOps Chat UI, advertise chat capability on its agent card:

spec = AgentSpec(
    agent_id="my-chat-agent",
    agent_card={
        "name": "My Chat Agent",
        "capabilities": {"chat": True, "ask": True, "streaming": True},
    },
)

Control plane chat sessions invoke the selected worker over A2A with input data that includes messages (conversation history), model, and prompt (latest user text). A2A message parts use the proto-style shape accepted by AgentOpsWorker, such as {"text": "..."} and {"data": {...}}.

Framework Adapters

LangChain

from komodor_agentops.langchain import KomodorCallbackHandler

handler = KomodorCallbackHandler(ops=transport_client, controlplane_run_id="run_1")
chain.invoke(input, config={"callbacks": [handler]})

Claude Code

Install hooks that forward Claude Code events to the controlplane:

install-cc-hooks --endpoint http://localhost:8000

SDK hooks for claude-agent-sdk:

from komodor_agentops.claude_code.sdk_hooks import agent_ops_hooks_for_run

hooks = agent_ops_hooks_for_run("run_1")

Architecture

komodor-agentops
  agentops-rpc, agentops-otel (wire types + OTel bootstrap)
  fastapi, uvicorn, a2a-sdk (worker/server runtime)
  httpx, pydantic, pydantic-settings, croniter, pyyaml, python-frontmatter
    |
    +-- [langchain]   -> langchain-core
    +-- [claude-code] -> claude-agent-sdk
    +-- [adk]         -> google-adk
    +-- [agno]        -> agno, anthropic
    +-- [server]      -> no-op alias (kept so existing [server] refs resolve)

Package Structure

src/komodor_agentops/
  __init__.py       # Public API
  _client.py        # Event buffer + flush
  _transport.py     # Full controlplane HTTP client
  _context.py       # TraceContext + ContextVar
  _events.py        # AgentOpsEvent dataclass
  _flush.py         # Async event batching
  _observe.py       # @observe() decorator
  _rpc.py           # JSON-RPC helpers
  _types.py         # Internal SDK types (AgentSpec, RunJob, etc.)
  worker.py         # AgentOpsWorker facade
  py.typed          # PEP 561 marker
  a2a/              # A2A protocol bridge
  stream/           # SSE/JSON-RPC streaming
  hooks/            # Reporter + heartbeat
  langchain/        # LangChain callback handler
  claude_code/      # Claude Code integration