Ad-hoc Step Recording: 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(
    kind: str | ActionKind | None = None,
    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,
    action_id: str | None = None,
    **kwargs,
) -> None

kind defaults to None and **kwargs is there only so that a missing or misspelled argument gets a clear message: in practice kind is required. A call without it raises TypeError: Missing required step argument: kind, and an unknown keyword raises TypeError: Unexpected step argument: <name>.

Parameters¶

Behaviour¶

• Requires an active task — set by start_task(), @kv.step(kind="task.start"), or a framework adapter such as KyvvuLangChainHandler. A handler owns the task it opens and refuses to start inside one that is already active, so with a handler you do not call start_task() yourself.
• A step.gate recorded with add_step has its properties.kind defaulted to human; kind must be human or guardrail, and a completed gate's output.decision must be pass or deny (ValueError otherwise).
• A task_id, action_id or step_name outside the identifier rules raises KyvvuValidationError before anything is recorded.
• 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. That is also why it works on a client built with Kyvvu(template=None) (no template at all), which @kv.step accepts too but start_task() refuses.
• Like @kv.step and the framework handlers, a block never ends the task: the blocked step is recorded (output.status is blocked), KyvvuBlockedError is raised, and the task stays open. The caller decides what happens next: error_task(), end_task(), or record an approval gate and retry in the same task. An open task nobody closes is evicted by the stale-task sweeper after KV_TASK_MAX_AGE_SECONDS (default 3600 s).

Usage examples¶

Standalone (manual task lifecycle)¶

from kyvvu import Kyvvu

kv = Kyvvu(api_url="...", api_key="...", agent_key="my-agent")
kv.register_agent(
    name="My Agent",
    purpose="Looks up customer records for the support team",
    # The step names of the step.resource / step.exec / step.credential
    # steps this agent records. The bundled baseline blocks any other.
    declared_tools=["fetch_user_data"],
)

kv.start_task()
kv.add_step(
    kind="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(kind="task.start")
def run_agent(prompt: str):
    result = call_llm(prompt)

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

    return result

Mixed with LangChain handler¶

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

kv = Kyvvu(api_url="...", api_key="...", agent_key="lc-agent")
kv.register_agent(
    name="LC Agent",
    purpose="Drafts and sends customer emails after human approval",
    declared_tools=["send_email"],  # the LangChain tool's name
)
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, and owns it:
    # do not call kv.start_task() yourself, and do not put @kv.step on this
    # tool (the handler already records it). add_step() inside the tool is fine:
    # use it to inject a gate the handler doesn't cover:
    approved = input(f"Approve sending to {recipient}? [y/n] ")
    kv.add_step(
        kind="step.gate",
        step_name="human_approval",
        input={"action": "send_email", "recipient": recipient},
        properties={"kind": "human"},
        output={"decision": "pass" if approved == "y" else "deny"},
    )
    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]})

A gate recorded with add_step carries its decision explicitly in output["decision"] ("pass" or "deny"), and its kind in properties["kind"] ("human" or "guardrail"). A decorated gate passes by returning and denies by raising kyvvu.GateDenied. See Python Decorator.

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 action kinds and verbs available
• Policies and Rules — how policy evaluation works