kv.add_step()

What you'll learn: How to record individual steps into the active task without wrapping a function in @kv.step.

kv.add_step() records a step into the active task imperatively. Use it for inline steps, third-party code you cannot decorate, dynamically chosen tools, or injecting a step.gate (human approval) inside a callback-based agent like LangChain. It is the missing primitive for mixing integration patterns cleanly — decorator, callback, and manual — within a single agent.

---

Method signature

kv.add_step(
    step_type: str | StepType,
    step_name: str | None = None,
    *,
    input: dict | None = None,
    output: dict | None = None,
    verb: str | Verb | None = None,
    properties: dict | None = None,
    meta: dict | None = None,
    task_id: str | None = None,
) -> None

---

Parameters

Parameter	Type	Default	Description
step_type	str | StepType	(required)	Step type enum or string (e.g. "step.gate", StepType.step_model).
step_name	str | None	None	Human-readable label. Defaults to the step_type value.
input	dict | None	None	Input data to record.
output	dict | None	None	Output data to record.
verb	str | Verb | None	None	HTTP-style verb ("GET", "POST", etc.).
properties	dict | None	None	Arbitrary key-value metadata.
meta	dict | None	None	Engine-reserved multi-agent correlation fields.
task_id	str | None	None	Explicit task ID. Falls back to the active task from start_task() or a framework adapter.

---

Behaviour

• Requires an active task — set by start_task(), @kv.step(step_type="task.start"), or a framework adapter such as KyvvuLangChainHandler.

• Evaluates policies before recording. Raises KyvvuBlockedError if the step is blocked by an active policy.

• Unlike @kv.step, no template matching is performed — the caller provides all fields explicitly.

• Unlike @kv.step, does not auto-call error_task() on block — the caller decides how to handle it.

---

Usage examples

Standalone (manual task lifecycle)

from kyvvu import Kyvvu

kv = Kyvvu(api_url="...", api_key="...")
kv.register_agent(agent_key="my-agent", name="My Agent", purpose="Demo")

kv.start_task()
kv.add_step(
    step_type="step.resource",
    step_name="fetch_user_data",
    verb="GET",
    input={"url": "https://api.example.com/users/42"},
    output={"status": 200, "user": {"name": "Alice"}},
)
kv.end_task()

Inside a @kv.step-decorated agent

@kv.step(step_type="task.start")
def run_agent(prompt: str):
    result = call_llm(prompt)

    # Inject an inline gate — no separate function needed
    kv.add_step(
        step_type="step.gate",
        step_name="safety_check",
        input={"content": result},
        output={"result": "pass", "check_type": "content_filter"},
    )

    return result

Mixed with LangChain handler

from langchain.tools import tool
from kyvvu import Kyvvu
from kyvvu.integrations.langchain import KyvvuLangChainHandler

kv = Kyvvu(api_url="...", api_key="...")
kv.register_agent(agent_key="lc-agent", name="LC Agent", purpose="Demo")
handler = KyvvuLangChainHandler(kv)

@tool
def send_email(recipient: str, body: str) -> str:
    """Send an email after human approval."""
    # The handler has already started a task via on_chain_start.
    # Use add_step() to inject a gate the handler doesn't cover:
    approved = input(f"Approve sending to {recipient}? [y/n] ")
    kv.add_step(
        step_type="step.gate",
        step_name="human_approval",
        input={"action": "send_email", "recipient": recipient},
        output={"result": "pass" if approved == "y" else "fail"},
        properties={"guard": {"check_type": "human_approval"}},
    )
    if approved != "y":
        return "Blocked by human reviewer."
    return do_send(recipient, body)

# Pass the handler when invoking the chain:
agent.invoke({"input": "..."}, config={"callbacks": [handler]})

---

get_current_task_id()

Convenience function that returns the task_id active in the current context (set by start_task() or a framework adapter), or None. Useful for advanced users who need to inspect the active task without a Kyvvu instance reference.

from kyvvu import get_current_task_id

task_id = get_current_task_id()  # str or None

---

Next steps

• Python Decorator — function-level step instrumentation with @kv.step

• LangChain / LangGraph — callback-based integration

• Atomic Behaviours — the step types and verbs available

• Policies and Rules — how policy evaluation works