LangChain / LangGraph¶

What you'll learn: How to instrument LangChain and LangGraph agents using the KyvvuLangChainHandler callback handler.

Setup¶

The same handler works for both LangChain and LangGraph — LangGraph uses langchain_core callbacks under the hood.

pip install "kyvvu-langchain" langchain

kyvvu-langchain installs only langchain-core. langchain provides create_agent, the current way to build a tool-calling agent, and installs LangGraph, which runs it. (LangGraph's older from langgraph.prebuilt import create_react_agent still imports in LangGraph 1.x, but calling it warns LangGraphDeprecatedSinceV10: it has moved to langchain.agents as create_agent.)

With the platform¶

The tool below returns a fixed example quote. Replace its body with your market-data provider when connecting a real agent.

from langchain.agents import create_agent
from langchain_core.tools import tool

from kyvvu import Kyvvu, KyvvuBlockedError
from kyvvu.schemas import Environment, RiskClassification
from kyvvu_langchain import KyvvuLangChainHandler


@tool
def search(ticker: str) -> str:
    """Return an example quote; replace with your market-data provider."""
    return f"{ticker}: 42.00 (example quote)"


kv = Kyvvu(
    api_key="KvKey-...",
    api_url="https://platform.kyvvu.com",
    agent_key="finance-agent",
    environment=Environment.DEVELOPMENT,
    risk_classification=RiskClassification.LIMITED,
    read_only_tools=["search"],
)
kv.register_agent(
    name="Finance Agent",
    purpose="Stock ticker lookup and portfolio analysis for internal use",
    # The person or team accountable for the agent.
    owner_id="ai-governance@example.com",
    declared_tools=["search"],
    metadata={"framework": "langchain", "tools": ["search"]},
)

# Any LangChain chat model. This one needs `pip install langchain-openai`
# and reads OPENAI_API_KEY.
agent = create_agent("openai:gpt-4o-mini", tools=[search])
handler = KyvvuLangChainHandler(kv)

try:
    result = agent.invoke(
        {"messages": [("user", "What is ACME trading at?")]},
        config={"callbacks": [handler]},
    )
    print(result["messages"][-1].content)
except KyvvuBlockedError as exc:
    print(f"Policy blocked: {exc}")

The handler is a pure adapter. It doesn't manage identity or registration; it is the same Kyvvu() + register_agent() pattern as the decorator integration. declared_tools matters: the bundled baseline that every registered agent gets blocks every tool call that is not on that list. So does owner_id: without it the baseline's "Agent must have a designated owner" policy warns at registration (medium, risk 0.50, not a block) and the platform opens an incident for it, as Your first agent explains. Without environment= an agent registers as KV_ENVIRONMENT says, and as production when that is unset; the snippet registers a development agent. A graph you build yourself with LangGraph's StateGraph takes the handler the same way (see LangGraph-specific notes).

Offline, with no account¶

No API key and no network are needed to enforce the bundled baseline or your own manifest. Give the client an agent_id, declare the names of the tools it may call, and pass the baseline as policies=. This script runs as written: a scripted chat model stands in for a real one, so it needs no provider key either.

"""Offline: a LangChain agent governed by the bundled baseline, no account."""
from langchain.agents import create_agent
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_core.tools import tool

from kyvvu import Kyvvu, KyvvuBlockedError
from kyvvu_baseline import load_baseline_policies
from kyvvu.schemas import RiskClassification
from kyvvu_langchain import KyvvuLangChainHandler


@tool
def search_orders(customer: str) -> str:
    """Look up a customer's open orders."""
    return f"{customer}: 2 open orders"


class ScriptedModel(GenericFakeChatModel):
    """Replays scripted messages; stands in for a real chat model."""

    def bind_tools(self, tools, **kwargs):  # create_agent binds the tools
        return self


model = ScriptedModel(messages=iter([
    AIMessage(content="", tool_calls=[
        {"name": "search_orders", "args": {"customer": "c-1"}, "id": "call-1"},
    ]),
    AIMessage(content="Customer c-1 has 2 open orders."),
]))

kv = Kyvvu(
    agent_id="orders-agent",
    risk_classification=RiskClassification.LIMITED,
    declared_tools=["search_orders"],
    read_only_tools=["search_orders"],
    policies=load_baseline_policies(),
    log_location="none",  # "stdout" (the offline default) prints the trace
)
agent = create_agent(model, tools=[search_orders])
handler = KyvvuLangChainHandler(kv)

try:
    result = agent.invoke(
        {"messages": [("user", "What is open for c-1?")]},
        config={"callbacks": [handler]},
    )
    print(result["messages"][-1].content)
except KyvvuBlockedError as exc:
    print(f"Policy blocked: {exc}")

Saved as agent.py and run with python agent.py, it prints (timestamps and warning locations omitted; task IDs shown as <task-id>):

kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO policy violation: task=<task-id> step='agent_response' action=warn risk=0.75 (2 violations)
UserWarning: Policy warning on 'agent_response': risk_score=0.75
kyvvu_engine.io.runner INFO task <task-id> ended — 7 steps (task.start×1, step.message×2, step.model×2, step.resource×1, task.end×1)
Customer c-1 has 2 open orders.

The declared tool ran and the agent answered. The warning is explained below.

You can set KV_DECLARED_TOOLS=search_orders instead of passing declared_tools= to Kyvvu(). The constructor argument wins over the environment variable; register_agent(declared_tools=...) wins over either when registration is used. Undeclared tool calls are blocked by the baseline. Separately, read_only_tools=["search_orders"] classifies the tool's callback steps as GET; this option is constructor-only and has no environment variable.

For a manifest of your own, set KV_POLICY_FILE=my-manifest.yaml (comma-separated for several files) instead of passing policies=, or pass Kyvvu(settings=KyvvuSettings(policy_file="my-manifest.yaml"), agent_id=...) with KyvvuSettings imported from kyvvu_engine.

Other offline settings:

• agent_id is required. Kyvvu(agent_key="x") alone fails on the first callback with KyvvuConfigError: no agent_id, because the only other way to get an id is register_agent(), which needs an API key. agent_key is not needed offline.
• The trace defaults to stdout; set KV_LOG_LOCATION (or log_location=) to a file path to keep it (KV_LOG_FORMAT=json for JSON lines), or to none as above.
• A policy source is required. Pass policies=, set KV_POLICY_FILE, or configure platform credentials; a client with none raises KyvvuConfigError on first use.
• Registration-time policies in the baseline are inert when nothing registers; step policies still enforce.

Why every run warns on agent_response¶

At the end of the top-level chain the handler records the agent's final answer as a step.message POST named agent_response. Under the bundled baseline that step is a compliance finding on every run: two EU AI Act Article 50 policies warn (high, risk 0.75), because the model's answer reaches a person with no gate recorded after the model call. The content is not marked as AI-generated (Art. 50(2)), and the person was not told they are dealing with an AI (Art. 50(1)). It is a warning, not a block: the answer is returned, the verdict is recorded on the step (meta["kyvvu.eval"].action is warn), and with the platform each violated policy opens an incident. The same warning on a decorator agent is explained in Your first agent.

The remedy is a passing step.gate after the last model call: the point where your application marks the content and discloses that it comes from an AI. With create_agent, an after_model middleware is that point. Add it to the script above:

from langchain.agents.middleware import after_model


@after_model
def mark_and_disclose(state, runtime):
    """Mark the final answer and record the Art. 50 checkpoint."""
    message = state["messages"][-1]
    if not message.tool_calls:
        notice = "\n\n(This answer was generated by an AI system.)"
        if isinstance(message.content, str):
            message.content += notice
        else:
            message.content = [*message.content, {"type": "text", "text": notice}]
        kv.add_step(
            "step.gate", "ai_disclosure",
            properties={"kind": "guardrail"},
            output={"decision": "pass"},
        )
        return {"messages": [message]}


agent = create_agent(model, tools=[search_orders], middleware=[mark_and_disclose])

The run then prints no policy violation line and no UserWarning, and the task records the gate: task ... ended — 8 steps (task.start×1, step.message×2, step.model×2, step.resource×1, step.gate×1, task.end×1). kv.add_step() needs no task_id here, because the middleware runs inside the task the handler opened.

Callback event mapping¶

The handler translates LangChain callback events to Kyvvu Behaviors. Each Behavior's kind field identifies its action kind. The built-in langchain template defines these mappings (rules are first-match-wins; order matters):

Model callbacks set properties.model to the invoked model id, add properties.provider when LangChain supplies ls_provider, and place generation settings under properties.model_params.

Tool callbacks record POST unless their exact name is in read_only_tools on the client. write_tools explicitly classifies writes. Both options are constructor-only; neither has a KV_ variable. A tool with data.* or target.trust properties must appear in one of these lists or its callback raises KyvvuConfigError. CrewAI declarations use the framework's sanitized tool name. These tool verbs take precedence over a custom template's verb. A custom template replaces the built-in (see below), and must also contain the task boundary rules. A callback the handler treats as a step that the template has no rule for is recorded as step.unknown, named after the event, and a warning is logged once per event. It is never dropped.

A tool is step.resource, not step.message¶

A LangChain tool is recorded as a step.resource, whatever it does. An undeclared send_email tool records POST; the action kind is always step.resource unless your own template says otherwise. It appears in the trace as:

step.resource POST send_email properties={"target":{"resource_type":"tool"}} action=allow

and a policy that targets step.message does not see it. A manifest written for the decorator, where @kv.step(ActionKind.step_message, Verb.POST) makes the send a step.message, therefore never fires on the same tool under the handler. What it does see is the handler's own step.message steps: user_input (GET) and, at the end of the chain, agent_response (POST). A "no outbound message after a PII read" policy then blocks the final answer after the email has already gone out, which is the wrong step at the wrong time.

The kind and properties come from the template, so customize it there. Copy the built-in template with the kyvvu command from kyvvu-cli (kyvvu template show langchain > org_langchain_template.yaml), and add these rules before tool_default. The first classifies the CRM tool as personal data, the second maps every send_* tool to a message send. Declare read_crm in read_only_tools and send_email in write_tools on the client:

- id: crm_tool
  description: CRM reads return personal data
  match:
    event: "on_tool_start"
    name_pattern: "^read_crm$"
  behavior:
    kind: "step.resource"
    verb: "GET"
    step_name: "{{ name }}"
    properties:
      data:
        classification: pii

- id: send_tool
  description: Tools named send_* send a message
  match:
    event: "on_tool_start"
    name_pattern: "^send_"
  behavior:
    kind: "step.message"
    verb: "POST"
    step_name: "{{ name }}"

Load it with KyvvuLangChainHandler(kv, template=BehaviorTemplate.from_path("org_langchain_template.yaml")) (see below). Run against a LangGraph agent with a fake model that calls read_crm and then send_email, with the taint policy named "Block external send after sensitive read". The trace shows the CRM read allowed and the send blocked before its body runs:

step.resource GET read_crm properties={"data":{"classification":"pii"}} action=allow
step.message POST send_email properties={} action=block

Without the template, even when the tool carries the PII label as metadata["kyvvu"], the email is allowed and the block arrives one step late, on agent_response, after the email has gone out.

Two limits of what the handler knows about a tool call:

• The tool's arguments arrive as one string, in input: {"input": "{'recipient': 'customer@example.com', 'body': 'hi'}"}. There is no message.recipient property, because the template context has only the tool's name, not its arguments. A recipient policy therefore matches the string, with field: input.input (see Field and property paths); a fixed classification can instead travel with the tool as metadata["kyvvu"] (below).
• agent_response is a step.message POST too. A policy on step.message POST also covers it. Scope a policy to sends by adding a step_name condition (field_matches_regex on step_name, pattern ^send_), as here, which blocks a send_* tool whose recipient is not on an approved domain:

name: LangChain recipient policy
version: "1.0.0"
policies:
  - name: Email only to approved domains
    rule_type: not
    params:
      condition:
        rule_type: all_of
        params:
          conditions:
            - rule_type: current_is
              params: {kind: step.message, verb: POST}
            - rule_type: field_matches_regex
              params: {field: step_name, pattern: "^send_"}
            - rule_type: not
              params:
                condition:
                  rule_type: field_matches_regex
                  params:
                    field: input.input
                    pattern: "'recipient': '[^']*@(example\\.com|partner\\.example)'"
    severity: critical
    enforcement_point: step_execution

Runnable offline example: block a send_* tool by recipient¶

This is the whole loop with no account, no API key, no network and no real model: a fake chat model that calls read_crm and then send_email, the two template rules above, and the recipient policy. Run it in a virtualenv (it was run as written with langchain 1.4, langgraph 1.2 and langchain-core 1.6; langchain is the package that provides create_agent):

pip install "kyvvu-langchain" langchain

"""Offline: a send_* tool becomes a step.message POST and a recipient policy blocks it."""
import subprocess
import sys

import yaml

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool
from langchain.agents import create_agent

from kyvvu import Kyvvu
from kyvvu_langchain import KyvvuLangChainHandler
from kyvvu.schemas import Environment, RiskClassification
from kyvvu_engine import KyvvuBlockedError
from kyvvu_engine import KyvvuSettings
from kyvvu_templates import BehaviorTemplate



# 1. Two tools. Neither knows anything about Kyvvu.
@tool
def read_crm(customer: str) -> str:
    """Look a customer up in the CRM."""
    return f"{customer}: Jane Doe, jane@example.com"


@tool
def send_email(recipient: str, body: str) -> str:
    """Send an email."""
    return f"sent to {recipient}"


# 2. A fake chat model: it asks for read_crm, then send_email, then answers.
#    The agent calls bind_tools() on its model, so the fake must implement it.
class FakeToolModel(BaseChatModel):
    script: list = []
    step: int = 0

    @property
    def _llm_type(self) -> str:
        return "fake-tool-model"

    def bind_tools(self, tools, **kwargs):  # create_agent calls this
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        msg = self.script[min(self.step, len(self.script) - 1)]
        self.step += 1
        return ChatResult(generations=[ChatGeneration(message=msg)])


def call(name, args, id):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": id}])


model = FakeToolModel(script=[
    call("read_crm", {"customer": "c-1"}, "1"),
    call("send_email", {"recipient": "attacker@evil.test", "body": "hi"}, "2"),
    AIMessage(content="done"),
])

# 3. The built-in langchain template plus two rules, inserted before tool_default.
#    Rules are first match wins, so order matters.
RULES = '''\
# docs:start tool-rules
  - id: crm_tool
    description: CRM reads return personal data
    match:
      event: "on_tool_start"
      name_pattern: "^read_crm$"
    behavior:
      kind: "step.resource"
      verb: "GET"
      step_name: "{{ name }}"
      properties:
        data:
          classification: pii

  - id: send_tool
    description: Tools named send_* send a message
    match:
      event: "on_tool_start"
      name_pattern: "^send_"
    behavior:
      kind: "step.message"
      verb: "POST"
      step_name: "{{ name }}"
# docs:end tool-rules

'''
base = subprocess.run(["kyvvu", "template", "show", "langchain"],
                      capture_output=True, text=True, check=True).stdout
assert "  - id: tool_default" in base
with open("org_langchain_template.yaml", "w") as f:
    f.write(base.replace("  - id: tool_default", RULES + "  - id: tool_default", 1))

# 4. One policy. The tool's arguments arrive as ONE string in `input`
#    ("{'recipient': '...', 'body': '...'}"), so the recipient rule reads `input.input`.
MANIFEST = r'''
# docs:start recipient-policy
name: LangChain recipient policy
version: "1.0.0"
policies:
  - name: Email only to approved domains
    rule_type: not
    params:
      condition:
        rule_type: all_of
        params:
          conditions:
            - rule_type: current_is
              params: {kind: step.message, verb: POST}
            - rule_type: field_matches_regex
              params: {field: step_name, pattern: "^send_"}
            - rule_type: not
              params:
                condition:
                  rule_type: field_matches_regex
                  params:
                    field: input.input
                    pattern: "'recipient': '[^']*@(example\\.com|partner\\.example)'"
    severity: critical
    enforcement_point: step_execution
# docs:end recipient-policy
'''
POLICIES = yaml.safe_load(MANIFEST)["policies"]

# 5. Offline: policies= is the policy source; no API key, no network.
kv = Kyvvu(
    settings=KyvvuSettings(agent_key="demo", environment="development", policies=POLICIES, log_location="none"),
    agent_id="demo-agent",
    environment=Environment.DEVELOPMENT,
    risk_classification=RiskClassification.LIMITED,
    read_only_tools=["read_crm"],
    write_tools=["send_email"],
)
handler = KyvvuLangChainHandler(
    kv, template=BehaviorTemplate.from_path("org_langchain_template.yaml")
)

agent = create_agent(model, [read_crm, send_email])
try:
    agent.invoke({"messages": [("user", "mail the customer")]}, config={"callbacks": [handler]})
    print("ALLOWED")
except KyvvuBlockedError as exc:
    print("BLOCKED:", exc)
    sys.exit(0)

Two things to notice. The template rule send_tool turns send_email into a step.message POST, which is what lets a policy on step.message see it (a tool is otherwise a step.resource). And the recipient rule reads input, because the tool's arguments arrive as one string, "{'recipient': 'attacker@evil.test', 'body': 'hi'}", not as a property.

The last line it prints is the decision. The recipient is not on example.com, so the step is blocked before the tool runs:

BLOCKED: Step 'send_email' blocked (risk_score=1.00): Email only to approved domains

Above that line you will also see LangChain's Error in KyvvuLangChainHandler.on_tool_start callback: KyvvuBlockedError(...) at WARNING, which is expected (see Block propagation). Change the recipient in the fake model's script to jane@example.com and the run prints ALLOWED. Run kyvvu template show langchain first if the command is not on your path: the script calls it to get the built-in template it extends.

Nested chain resolution¶

LangChain/LangGraph agents often produce deeply nested chain callbacks. The handler resolves nested chains to keep the behavioral trace readable:

• Top-level chain events produce Behaviors.
• Deeply nested internal chains are collapsed unless they represent distinct logical steps.
• Tool calls within chains are always surfaced.

Block propagation¶

When a policy blocks a step during a LangChain callback:

1. The engine records the blocked step (output.status is blocked, with meta["kyvvu.eval"]) and the handler raises KyvvuBlockedError.
2. With raise_error=True (default), the error propagates up through the LangChain callback chain, halting execution.
3. LangChain logs the block at WARNING, for example Error in KyvvuLangChainHandler.on_tool_start callback: KyvvuBlockedError("Step 'send_email' blocked (risk_score=1.00): ..."). That line is expected: it is LangChain reporting the exception the handler raised on purpose, not a second failure. Your code gets the same KyvvuBlockedError from invoke().
4. LangChain then reports the failure through its own on_chain_error callback, which closes the task the handler opened as task.error. The block itself never ends the task: if you catch the error inside the chain (for example in a tool) and carry on, the task is still open.
5. The agent code can catch KyvvuBlockedError at the top level.

from kyvvu import KyvvuBlockedError

handler = KyvvuLangChainHandler(kv)
try:
    result = agent.invoke(query, config={"callbacks": [handler]})
except KyvvuBlockedError as exc:
    print(f"Policy blocked: {exc}")

When a policy warns instead of blocking¶

A violated policy of severity high, medium or low does not stop the step. The handler lets it run, records the verdict in the trace (meta["kyvvu.eval"].action is warn), logs policy violation: ... action=warn at INFO, and raises a Python UserWarning (Policy warning on 'send_email': risk_score=0.75). The warning names the line in your code that started the run, for example your agent.invoke(...) call, not a line inside Kyvvu. Python shows it once per location; filter or escalate it with the warnings module (warnings.simplefilter("error", UserWarning) turns a warn into an exception, which is a quick way to see where in a test a policy warns).

Task lifecycle¶

The handler owns the task it opens, and manages the whole lifecycle: - A task starts when the top-level chain starts (on_chain_start). Nested chains open no task. - The task ends when the top-level chain completes (on_chain_end). - If the chain raises an exception, on_chain_error closes it as task.error.

Because the handler opens the task, do not call kv.start_task() around it. If a task is already active in the current context when the handler tries to start one, it raises KyvvuConfigError telling you to drop the explicit start_task() (and, if an earlier step was blocked and left its task open, to end that task first with kv.end_task() or kv.error_task()). The handler closes only the tasks it opened: a task you ended yourself mid-chain gets no second task.end.

The decorator and the handler build steps differently. The decorator records a step from its own arguments when its template has no rule for it, whereas an unmatched callback event is recorded as step.unknown.

Do not decorate a tool the handler runs¶

The handler already records every tool it runs, so putting @kv.step on the tool function itself would record the one call twice. That raises KyvvuConfigError when the decorated tool is called:

'send_email' is a tool the framework handler is already running and recording, so @kv.step on it would record the call twice. Remove the @kv.step from the tool (or exclude the tool from the handler). Decorate only functions the framework cannot see, such as a helper the tool calls. If this is a different function that happens to share the tool's name, rename the helper.

A decorated helper that a tool calls, and kv.add_step(...) inside a tool (for example an approval gate), are both fine. Two limits of the check: it compares the function's name with the name of the tool that is running, so a helper that happens to share a tool's name is refused and should be renamed; and it relies on Python context variables, so it sees a tool only in the context that runs it (code that starts a fresh context, such as a plain threading.Thread, is not checked and is recorded normally).

Attaching Kyvvu attributes to tools: two layers¶

Out of the box, the built-in langchain template gives every step a correct kind and verb. To drive richer policies you often want extra properties on a step — a data classification, a target system, a trust level. There are two first-class ways to attach them, and they compose.

Org-wide behavior template (base layer)¶

A behavior template is a shared YAML file (.btmpl / .yaml) that maps tools by name-pattern to Kyvvu properties. Give it to the handler with the template= constructor kwarg, or point every integration built with the default at it with the KV_TEMPLATE_LOCATION environment variable. A custom template replaces the built-in langchain template; it does not extend it, so start from kyvvu template show langchain > org_langchain_template.yaml and edit the copy. It must also contain the task boundary rules (on_chain_start, on_chain_end, on_chain_error), or the handler refuses to build. Because it lives in one file, it is the natural home for fleet-wide policy: one team owns it, every agent that loads it inherits the same baseline, and you can add tool properties without changing the tool implementation. Tool verbs still need client declarations.

from kyvvu_templates import BehaviorTemplate
from kyvvu_langchain import KyvvuLangChainHandler

template = BehaviorTemplate.from_path("org_langchain_template.yaml")
handler = KyvvuLangChainHandler(kv, template=template)

Kyvvu(..., template=...) would not do this: that template is for the decorator and start_task(), not for the handler.

This rule, added to the copy before the tool_default rule, classifies any tool named get_invoice as financial data. It matches on the tool name and sets the same property a tool could set about itself:

Declare get_invoice in read_only_tools on the client so this classified read records GET.

rules:
  - id: invoice_tool
    description: Invoice lookups handle financial data
    match:
      event: "on_tool_start"
      name_pattern: "^get_invoice$"
    behavior:
      kind: "step.resource"
      verb: "GET"
      step_name: "{{ name }}"
      properties:
        data:
          classification: financial

Rules are first-match-wins. Nothing falls through to the built-in langchain template: an event your template has no rule for is recorded as step.unknown.

Per-tool metadata["kyvvu"] (local attributes)¶

A tool can also declare its own Kyvvu attributes, co-located with the tool definition, under the reserved metadata["kyvvu"] key. The value is shaped exactly like a step's properties:

from langchain_core.tools import Tool

get_invoice = Tool(
    name="get_invoice",
    func=fetch_invoice,
    description="Fetch an invoice by ID",
    metadata={"kyvvu": {"data": {"classification": "financial"}}},
)

This produces the same data.classification: financial property as the org template rule above — but it travels with the tool, which is ideal for tools you own or for local nuance the fleet template shouldn't have to know about. The handler forwards metadata["kyvvu"] for tools, retrievers, and models (on_tool_start, on_retriever_start, on_llm_start / on_chat_model_start).

Precedence¶

Metadata deep-merges over template properties at each leaf, preserving template siblings. Tool metadata wins on conflicts. For example, if the org template classifies get_invoice as internal and the tool declares financial, the emitted step records financial and logs a deduped conflict warning. For step.model, the invoked model and framework-reported provider remain authoritative; run metadata cannot relabel them.

When to use which¶

• Org template — fleet-wide policy, consistent classification across many agents, and tools you don't own (third-party or vendor tools). One file, one owner.
• Per-tool metadata — tools you own, or local attributes that belong next to the tool definition, including overrides of template values.

Framework tags¶

Framework tags on a run are forwarded into properties.tags, minus internal LangChain/LangGraph bookkeeping tags (those prefixed seq:, graph:, or langsmith:). With no metadata and no tags, step properties are unchanged from the template match.

LangGraph-specific notes¶

LangGraph agents use the same handler. This complete offline example passes it via config and governs a declared echo tool:

from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

from kyvvu import Kyvvu
from kyvvu_baseline import load_baseline_policies
from kyvvu_langchain import KyvvuLangChainHandler


class State(TypedDict):
    message: str


@tool
def echo(message: str) -> str:
    """Return the input for this offline example."""
    return f"Received {message}"


def respond(state: State, config: RunnableConfig) -> State:
    return {"message": echo.invoke(state["message"], config=config)}


graph = StateGraph(State)
graph.add_node("respond", respond)
graph.add_edge(START, "respond")
graph.add_edge("respond", END)
app = graph.compile()

kv = Kyvvu(
    agent_id="echo-graph",
    policies=load_baseline_policies(),
    declared_tools=["echo"],
    read_only_tools=["echo"],
    log_location="none",
)
input_data = {"message": "Hello"}
handler = KyvvuLangChainHandler(kv)
result = app.invoke(input_data, config={"callbacks": [handler]})
print(result["message"])

The behavioral trace stays flat — there are no node-transition pseudo-steps. Tool calls within nodes appear as step.resource events, and (with KyvvuLangGraphHandler) each step records its originating node under properties.langgraph.node_name.

Dedicated LangGraph handler¶

For richer graph-aware tracing, use KyvvuLangGraphHandler (from kyvvu_langchain), a subclass of KyvvuLangChainHandler that adds two things:

• Node metadata — injects properties.langgraph.node_name from the callback metadata, so the trace records which graph node each step ran in.
• Interrupt mapping — maps LangGraph's GraphInterrupt (raised by interrupt()) to a step.gate behavior, so human-in-the-loop breakpoints are governable by gate policies.

from kyvvu_langchain import KyvvuLangGraphHandler

handler = KyvvuLangGraphHandler(kv)
result = app.invoke(input_data, config={"callbacks": [handler]})

The shared KyvvuLangChainHandler still works for LangGraph; the dedicated handler is opt-in when you want node metadata or HITL gate mapping.

Parallel tool calls within a step¶

When the LLM emits multiple tool_use blocks and the runtime dispatches them concurrently (e.g., asyncio.gather in LangGraph), Kyvvu sees N sibling tool calls under the same parent run_id. Each step's timestamp is when the handler saw the call start. Each step also gets a seq, which is strictly increasing per (agent_id, task_id) and may have gaps, but says nothing about which of two concurrent siblings happened first. (The runner currently assigns it when a step is recorded; that is not a promise.) Neither field implies causality between concurrent siblings. The task is written by the handler that opened it.

Path policies operate across steps, not within them. A policy like "must call X before Y" is well-defined when X and Y are in different steps (X in step 1, Y in step 2). It is ill-defined when X and Y are dispatched concurrently within the same step — their ordering is non-deterministic. Policy authors should specify ordering constraints across steps, not between parallel calls within one.

Kyvvu evaluates the path; it does not serialise the dispatcher. If a policy requires sequential execution, the developer must enforce it in the runtime — sequential awaits, or a single-tool-per-step constraint in the agent prompt. Kyvvu records what happened; it does not control dispatch order.

Integration status¶

CrewAI¶

Install the CrewAI listener package in the agent environment:

pip install kyvvu-crewai

kyvvu-crewai installs CrewAI and the kyvvu Python client together, into one environment, and they do not conflict. It needs Python 3.11 to 3.13, because CrewAI does not support Python 3.14 yet. Its range is crewai>=1.15.21,<2 (<2 is the next major), and 1.15.32 is the release Kyvvu tests. See CrewAI for the setup and a runnable offline example.

The listener uses the crewai template and follows the same rules as the LangChain handler: a custom template replaces the built-in and must cover crew_kickoff_started, crew_kickoff_completed and crew_kickoff_failed; the listener owns the crew's task and refuses to start inside an active one. Synchronous CrewAI before-call hooks evaluate tools and model calls before execution. A block stops the call and aborts the crew; the engine records the blocked step and the listener closes the task with task.error. The listener refuses to start if these hooks are unavailable. CrewAI's event bus still carries task boundaries and messages; listener.blocked_exception retains the original policy block for callers.

Interested in a framework not listed here? Contact us at support@kyvvu.com.

Next steps¶

• Python Decorator — alternative integration for custom Python agents
• Writing a New Integration — build a handler for another framework
• Templates — customize the event-to-Behavior mapping