meander-agent

Typed Python client for the plan-sending side of meander.

Your agent records what it did as a traceable claim with provenance, and asks exactly one question of a rule in your ontology. This package turns that into a meander.plan, checks its shape, and attaches it to a real OpenTelemetry span. Your OTLP export carries the span to meander. The client never imports meander and holds no ontology.

The model in one paragraph

A meander.plan is an intended action plus a claim plus one question, not a decision:

• action: the one thing the agent wants to do, declared up front (kind, name, description, optional target: {entity, identity}). This is the agent's statement of intent — meander shows it as "what the agent wants". It is required, but it is never written to the fact graph and never changes the verdict; the server stores it as part of the plan.
• facts: what the agent observed or judged. Each fact carries its own provenance (origin.kind: tool | api | human | agent). A tool result is tool; the agent's own judgement is agent.
• relations: claimed relationships between entities (optional).
• question: exactly one question for a rule (derivation) that already exists in your world, for example "does requires_review hold here?". The agent does not decide. Whether the rule fires is decided by meander on the server, and a human reviews it.

The client checks only the shape: required fields, types, one required action, exactly one question, the origin vocabulary, and plan_version == 1. Whether the entities, properties, and derivations really exist is known only to the server, because only the server has your ontology. That is why this package does not import meander and loads no ontology.

Guarantees (never a silent no-op): a failed run never attaches an attribute; a span that can no longer be written refuses the attribute out loud (AttachResult(ok=False, span_not_recording)); a plan with a bad shape is reported with a field path instead of being written half-way.

Installation

pip install meander-agent            # transport + emitter (slim, no LLM SDK)
pip install "meander-agent[claude]"  # + Claude Agent SDK binding (not live-validated for 0.2.2 yet, see §3)
pip install "meander-agent[openai]"  # + OpenAI Agents SDK binding (live-validated for 0.2.2)

1. Declare your vocabulary

You give the agent the names of your world, so it invents nothing. For each entity its identity fields and properties, for each relation its endpoints, plus the derivation IDs it may ask about:

vocabulary = {
    "entities": {
        "Order":   {"identity": ["order_id"], "properties": ["amount", "risk"]},
        "Case":    {"identity": ["case_id"],  "properties": []},
        "Outcome": {"identity": ["outcome_id"], "properties": []},
    },
    "relations": {
        "concerns": {"from": "Case", "to": "Order"},
    },
    "derivation_ids": ["requires_review.high_value"],
}

From this the package builds the prompt fragment (it names only these names to the model) and the JSON schema for the structured output (it allows only these names).

2. Set up transport

If you have no OTel of your own, one call is enough. endpoint is the full OTLP traces URL of your source, source_key is that source's bearer token:

from meander_agent import init_meander

client = init_meander(
    endpoint="https://<host>/api/sources/<source_id>/v1/traces",
    source_key="<bearer-token>",
)
# client.tracer  -> the wired OTel tracer
# client.shutdown() / client.force_flush()  -> finish the export

init_meander sets no global provider; the client keeps its own. If you already have an OTel setup, skip init_meander and pass your tracer directly (see section 5).

3. Run an agent (Claude)

Status: not live-validated for meander-agent 0.2.2. The Claude binding's provider live smoke has not yet passed against 0.2.2 (external billing gap on the Anthropic key), so this binding is not recommended for customer use until that smoke is green once. The OpenAI binding (§3b) is live-validated for 0.2.2 end-to-end, so prefer it until then.

from meander_agent.claude import run_with_plan

result = run_with_plan(
    "Handle order ORD-42. Call the usual tools, claim the facts you gathered "
    "with their provenance, and ask exactly one question of the derivation "
    "requires_review.high_value. Do NOT make a decision.",
    vocabulary=vocabulary,
    tracer=client.tracer,
)
print("attached" if result.attached else f"no plan set: {result.error_state}")
client.shutdown()   # export the span

The binding opens the root span, runs the model with structured output, locks on error or abort, and on success attaches the checked plan. It returns a RunResult (see section 4). Needs the [claude] extra and an ANTHROPIC_API_KEY.

3b. The same run over OpenAI (official SDK pattern)

The [openai] extra binds the OpenAI Agents SDK. You write ordinary agent code (Agent, Runner.run, final_output_as); meander only supplies the prompt fragment and the structured output_type, and finalizes the run:

import os
from agents import Agent, Runner
from meander_agent import init_meander, meander_run
from meander_agent.openai import PlanModel, plan_output_type

vocabulary = {
    "entities": {"Answer": {"identity": ["id"], "properties": ["text"]}},
    "relations": {},
    "derivation_ids": ["answer.acceptable"],
}

client = init_meander(
    endpoint=os.environ["MEANDER_OTLP_ENDPOINT"],
    source_key=os.environ["MEANDER_SOURCE_KEY"],
)

with meander_run(vocabulary=vocabulary, tracer=client.tracer) as run:
    agent = Agent(
        name="History Tutor",
        instructions="You answer history questions clearly and concisely.\n\n" + run.prompt_fragment,
        output_type=plan_output_type(PlanModel),
    )
    result = await Runner.run(agent, "When did the Roman Empire fall?")
    plan = result.final_output_as(PlanModel, raise_if_incorrect_type=True)
    meander_result = run.finalize(plan.model_dump(by_alias=True), None)

client.shutdown()
print(meander_result.plan)

If you would rather not write that glue, the binding ships the one-liner run_with_plan(task, *, vocabulary, tracer), which wraps the same Runner.run -> final_output_as -> run.finalize chain for you (async variant: run_with_plan_async):

from meander_agent.openai import run_with_plan

result = run_with_plan(task, vocabulary=vocabulary, tracer=client.tracer)

In a Jupyter notebook, use await Runner.run(...) directly; do not wrap it in asyncio.run(...), because the notebook already runs an event loop.

4. What you get back

RunResult:

• attached: bool tells whether meander.plan was attached to the span.
• plan: dict | None is the attached plan (on success).
• shape_errors: list[ShapeError] lists shape errors with path / code / message, when the output failed the shape check.
• error_state is None on success; otherwise it is the run's failure exit (SDK error, abort, type mismatch). When it is set, nothing is ever attached.

So no meander.plan always means one of two things: a failed run (error_state) or an output with a bad shape (shape_errors). Both are in the result; nothing disappears silently.

5. Your own tracing / your own SDK (the core)

If you want to wire your own SDK (or use your own tracer), you drive the SDK-neutral context manager yourself. It hands you the fragment and the schema, and takes care of parsing, the shape check, locking, and attaching:

from meander_agent import meander_run

with meander_run(vocabulary=vocabulary, tracer=my_tracer) as run:
    # run.prompt_fragment : shape + allowed vocabulary -> instruction to your SDK
    # run.output_schema   : JSON schema, if your SDK can do structured output
    output, error = call_your_llm(task, instructions=run.prompt_fragment,
                                  schema=run.output_schema)
    # output: the structured result (dict) OR a plain JSON string.
    # error_state: None on success, otherwise any detail (=> lock).
    result = run.finalize(output, error_state=error)

The core never passes prompts to the SDK itself and never reads SDK results; that is your binding's job. The bundled Claude and OpenAI bindings work the same way.

6. A deterministic plan without an LLM

If you build the plan yourself (tests, rule-based agents, an auth-free path), you use the emitter directly:

from meander_agent import attach_plan, validate_plan_shape

plan = {
    "plan_version": 1,
    "action": {
        "kind": "tool_call",
        "name": "issue_refund",
        "description": "Issue a refund for order ORD-42",
        "target": {"entity": "Order", "identity": {"order_id": "ORD-42"}},
    },
    "facts": [
        {"entity": "Order", "identity": {"order_id": "ORD-42"},
         "property": "amount", "value": 900,
         "origin": {"kind": "tool", "ref": "lookup_order"}},
    ],
    "relations": [
        {"relation": "concerns",
         "from": {"entity": "Case", "identity": {"case_id": "C-1"}},
         "to":   {"entity": "Order", "identity": {"order_id": "ORD-42"}},
         "origin": {"kind": "agent"}},
    ],
    "question": {
        "derivation_id": "requires_review.high_value",
        "subject": {"entity": "Case", "identity": {"case_id": "C-1"}},
    },
}

errors = validate_plan_shape(plan)            # pure shape check (empty = ok)
with client.tracer.start_as_current_span("agent.run") as span:
    res = attach_plan(span, plan)             # attaches if the shape is valid
    if not res.ok:
        print("not attached:", [e.as_dict() for e in res.errors])
client.shutdown()

Public surface

Development

pixi run test          # random order (pytest-randomly)
pixi run -- pytest -p no:randomly    # fixed order

The credential-gated live LLM tests (Claude/OpenAI) run where the keys are set, and are skipped otherwise. The deterministic suite always runs, with no mocks.