> For the complete documentation index, see [llms.txt](https://docs.kyvvu.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.kyvvu.com/reference/add-step.md). # 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 ```python 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: `. *** ## Parameters | Parameter | Type | Default | Description | | ------------ | --------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `kind` | `str \| ActionKind \| None` | `None` | Required in practice: `None` raises `TypeError`. Action kind enum or string (e.g. `"step.gate"`, `ActionKind.step_model`). | | `step_name` | `str \| None` | `None` | Human-readable label: 1-255 characters, no control characters. Defaults to the `kind` 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: letters, digits and `. _ : -`, 1-100 characters. Falls back to the active task from `start_task()` or a framework adapter. The task must be open in this client. | | `action_id` | `str \| None` | `None` | Identifier for this action: letters, digits and `. _ : -`, 1-255 characters. Generated when omitted. Unique within the task, except to pair an intended step with its own completion; a duplicate raises `ValueError`. | *** ## 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](/core-concepts/tasks.md#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) ```python 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 ```python @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 ```python 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](/integrations/decorator.md#gates-pass-on-a-normal-return). *** ## 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. ```python from kyvvu import get_current_task_id task_id = get_current_task_id() # str or None ``` *** ## Next steps * [Python Decorator](/integrations/decorator.md) — function-level step instrumentation with `@kv.step` * [LangChain / LangGraph](/integrations/langchain.md) — callback-based integration * [Atomic Behaviours](/core-concepts/behaviours.md) — the action kinds and verbs available * [Policies and Rules](/core-concepts/policies.md) — how policy evaluation works