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 — the whole of a worker's main()

A worker takes runs and returns results. It serves no HTTP and needs no inbound network access: it connects out to the control plane, is told when work is available, and claims it.

from pathlib import Path

from komodor_agentops import AgentOpsWorker, AgentSpec, Run

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

async def on_run(run: Run) -> dict:
    return {"answer": f"handled {run.input.get('prompt', '')}"}

def main() -> None:
    AgentOpsWorker(agent=spec, on_run=on_run).run()

run.input is the payload the run was created with; the dict you return becomes the run's output. Raise to fail the run — that is on_run's only failure channel, and the error is recorded on the run.

No host or port. Nothing dials your worker, so there is nothing to bind. A deployment needs egress to the control plane and no Service, Ingress or open port; the customer namespace chart denies inbound traffic by default.

A worker built against komodor-agentops 0.1.x may instead be using run_worker(agent, handler) with a handler taking an A2AMessage and returning an A2ATask, served on a port. That runtime is gone: it served an A2A app nothing dials any more. Move the handler to on_run= — it receives a Run and returns a mapping — and construct AgentOpsWorker from the package root as above. Run.input is the whole payload the message's parts used to carry, and the answer's human-readable half is the reserved text key rather than an artifact.

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 — the control plane materialises a routing rule per channel on heartbeat and dispatches matching messages; the agent is invoked when @mentioned in one of those channels):

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},
    },
)

A chat run arrives like any other run: run.input carries messages (the conversation history), model, and prompt (the latest user text). There are no A2A message parts to unpack — the control plane creates a run and your worker claims it, so a chat handler reads the same run.input dict as every other handler.

To stream the answer as it is produced rather than returning it whole, emit it through the ambient client; the control plane relays each new piece of the run's answer text to the browser as it appears.

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)
  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; the SDK
                         serves no HTTP, so there is nothing for it to pull in)

Package Structure

src/komodor_agentops/
  __init__.py       # Public API — the root is the API; submodule paths are internal
  worker_runtime.py # AgentOpsWorker: connect, register, claim, run, report
  worker_client.py  # The worker protocol, and the run scope
  messaging.py      # The transport's composition root (downlink + uplink)
  client.py         # Event buffer + flush
  py.typed          # PEP 561 marker
  transport/        # The byte-level connection (SSE, reconnect)
  bus/              # Frames: the envelope, the router, the uplink
  machinery/        # Delivery: batching, retry, sequencing
  core/             # Context, events, secrets, types, @observe
  worker/           # Run contract, liveness, hooks, channel listener
  adapters/         # Claude Code, ADK, Agno, LangChain, nanobot