kyvvu_langchain Python API¶

Kyvvu callback handlers for LangChain and LangGraph.

kyvvu_langchain.KyvvuLangChainHandler¶

KyvvuLangChainHandler(kyvvu: 'Kyvvu', template: object = default_template) -> None

LangChain callback handler for Kyvvu compliance monitoring.

Pure adapter: accepts an already-constructed Kyvvu instance and binds LangChain's callback stream to it via FrameworkAdapter.

raise_error = True causes LangChain to propagate exceptions raised inside callback methods (required for KyvvuBlockedError to abort the chain).

Parameters:

• kyvvu ('Kyvvu') – A configured Kyvvu instance.
• template (object) – A BehaviorTemplate (replaces the built-in and governs both steps and task boundaries), or the default default_template. None is refused.

Example::

kv = Kyvvu(api_url="...", api_key="KvKey-...", agent_key="bot")
kv.register_agent(name="Math Agent")
handler = KyvvuLangChainHandler(kv)
agent.invoke({"input": "2+2"}, config={"callbacks": [handler]})

Functions:

• build_properties – Enrich template properties with framework tags and tool metadata.
• on_agent_finish – Leave response evaluation and task closure to the root chain end.
• on_chain_end – Top-level chain end closes the task.
• on_chain_error – Top-level chain error emits task.error and flushes.
• on_chain_start – Top-level chain start opens a task; sub-chains are tracked for depth.
• on_chat_model_start – Evaluate step.model before chat model execution.
• on_custom_event – Handle application-defined custom events (adispatch_custom_event).
• on_llm_end – Record completed step.model.
• on_llm_error – Record failed step.model.
• on_llm_start – Evaluate step.model before LLM execution.
• on_retriever_end – Record completed retriever step.
• on_retriever_error – Record failed retriever step.
• on_retriever_start – Evaluate step.resource GET before retriever execution.
• on_tool_end – Record completed tool step.
• on_tool_error – Record failed tool step.
• on_tool_start – Evaluate step.resource before tool execution.

Attributes:

• boundary_contexts (Mapping[ActionKind, Sequence[Dict[str, Any]]]) –
• emitted_events (frozenset[str]) –
• raise_error (bool) –
• run_inline (bool) –
• template_name (str) –
• template_package (str) –

Bind LangChain callbacks to the client and behavior template.

kyvvu_langchain.KyvvuLangChainHandler.boundary_contexts¶

boundary_contexts: Mapping[ActionKind, Sequence[Dict[str, Any]]] = {ActionKind.task_start: ({'event': 'on_chain_start'},), ActionKind.task_end: ({'event': 'on_chain_end'},), ActionKind.task_error: ({'event': 'on_chain_error'},)}

kyvvu_langchain.KyvvuLangChainHandler.build_properties¶

build_properties(matched: Dict[str, Any], context: Dict[str, Any], step_run_id: str) -> Dict[str, Any]

Enrich template properties with framework tags and tool metadata.

Run metadata overlays template properties and framework tags at leaves. A model's invoked name and framework provider remain authoritative.

Parameters:

• matched (Dict[str, Any]) – The template-matched behavior field dict.
• context (Dict[str, Any]) – The original event context for this step (carries the LangChain metadata and tags forwarded by the callback).
• step_run_id (str) – The pairing run identifier for this step.

Returns:

• Dict[str, Any] – Properties dict to attach to the emitted Behavior.

kyvvu_langchain.KyvvuLangChainHandler.emitted_events¶

emitted_events: frozenset[str] = frozenset({'on_chain_start', 'on_chain_end', 'on_chain_error', 'on_message_in', 'on_message_out', 'on_llm_start', 'on_chat_model_start', 'on_retriever_start', 'on_tool_start', 'on_custom_event'})

kyvvu_langchain.KyvvuLangChainHandler.on_agent_finish¶

on_agent_finish(finish: AgentFinish, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Leave response evaluation and task closure to the root chain end.

kyvvu_langchain.KyvvuLangChainHandler.on_chain_end¶

on_chain_end(outputs: Dict[str, Any], *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Top-level chain end closes the task.

kyvvu_langchain.KyvvuLangChainHandler.on_chain_error¶

on_chain_error(error: BaseException, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Top-level chain error emits task.error and flushes.

kyvvu_langchain.KyvvuLangChainHandler.on_chain_start¶

on_chain_start(serialized: Dict[str, Any], inputs: Dict[str, Any], *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Top-level chain start opens a task; sub-chains are tracked for depth.

Raises:

• KyvvuBlockedError – If the task.start is blocked by policy.

kyvvu_langchain.KyvvuLangChainHandler.on_chat_model_start¶

on_chat_model_start(serialized: Dict[str, Any], messages: List[List[Any]], *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Evaluate step.model before chat model execution.

Raises:

• KyvvuBlockedError – If the policy engine blocks this model call.

kyvvu_langchain.KyvvuLangChainHandler.on_custom_event¶

on_custom_event(name: str, data: Any, *, run_id: UUID, parent_run_id: Optional[UUID] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

Handle application-defined custom events (adispatch_custom_event).

Maps to the template on_custom_event rule. Default behavior is step.unknown unless the template overrides by name pattern.

Parameters:

• name (str) – Event name chosen by the application.
• data (Any) – Arbitrary event payload.
• run_id (UUID) – LangChain run identifier.
• parent_run_id (Optional[UUID]) – Parent run identifier.
• tags (Optional[List[str]]) – Optional tags from the run context.
• metadata (Optional[Dict[str, Any]]) – Optional metadata from the run context.
• **kwargs (Any) – Additional callback fields supplied by LangChain.

kyvvu_langchain.KyvvuLangChainHandler.on_llm_end¶

on_llm_end(response: LLMResult, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record completed step.model.

kyvvu_langchain.KyvvuLangChainHandler.on_llm_error¶

on_llm_error(error: BaseException, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record failed step.model.

kyvvu_langchain.KyvvuLangChainHandler.on_llm_start¶

on_llm_start(serialized: Dict[str, Any], prompts: List[str], *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Evaluate step.model before LLM execution.

Raises:

• KyvvuBlockedError – If the policy engine blocks this LLM call.

kyvvu_langchain.KyvvuLangChainHandler.on_retriever_end¶

on_retriever_end(documents: Any, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record completed retriever step.

kyvvu_langchain.KyvvuLangChainHandler.on_retriever_error¶

on_retriever_error(error: BaseException, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record failed retriever step.

kyvvu_langchain.KyvvuLangChainHandler.on_retriever_start¶

on_retriever_start(serialized: Dict[str, Any], query: str, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Evaluate step.resource GET before retriever execution.

Raises:

• KyvvuBlockedError – If the policy engine blocks this retrieval.

kyvvu_langchain.KyvvuLangChainHandler.on_tool_end¶

on_tool_end(output: str, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record completed tool step.

kyvvu_langchain.KyvvuLangChainHandler.on_tool_error¶

on_tool_error(error: BaseException, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Record failed tool step.

kyvvu_langchain.KyvvuLangChainHandler.on_tool_start¶

on_tool_start(serialized: Dict[str, Any], input_str: str, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Evaluate step.resource before tool execution.

Raises:

• KyvvuBlockedError – If the policy engine blocks this tool call.

kyvvu_langchain.KyvvuLangChainHandler.raise_error¶

raise_error: bool = True

kyvvu_langchain.KyvvuLangChainHandler.run_inline¶

run_inline: bool = True

kyvvu_langchain.KyvvuLangChainHandler.template_name¶

template_name: str = 'langchain'

kyvvu_langchain.KyvvuLangChainHandler.template_package¶

template_package: str = 'kyvvu_langchain.templates'

kyvvu_langchain.KyvvuLangGraphHandler¶

KyvvuLangGraphHandler(kyvvu: 'Kyvvu', template: object = default_template, extract_tools: bool = False, graph: Any = None) -> None

LangGraph callback handler with graph-specific enrichments.

Extends KyvvuLangChainHandler to:

1. Inject properties.langgraph.node_name from callback metadata.
2. Map GraphInterrupt to step.gate (HITL breakpoints).
3. Optionally auto-extract declared_tools from a ToolNode.

The behavioral trace remains flat — no node-transition pseudo-steps.

Parameters:

• kyvvu ('Kyvvu') – A configured Kyvvu instance.
• template (object) – A BehaviorTemplate (replaces the built-in and governs both steps and task boundaries), or the default default_template. None is refused.
• extract_tools (bool) – When True and a graph is provided, inspect the graph's ToolNode for bound tools and pass them to register_agent(declared_tools=[...]).
• graph (Any) – Optional compiled LangGraph graph used for tool extraction.

Example::

kv = Kyvvu(api_url="...", api_key="KvKey-...", agent_key="bot")
kv.register_agent(name="Agent")
handler = KyvvuLangGraphHandler(kv, extract_tools=True, graph=compiled)

Functions:

• build_properties – Inject langgraph.node_name into the step properties.
• emit_task_start – Open the task, adding properties.langgraph thread identifiers.
• on_chain_error – Intercept GraphInterrupt and emit step.gate instead of task.error.
• on_chain_start – Chain start; a root run records its checkpointer thread on task.start.
• on_chat_model_start – Chat model start with node metadata capture.
• on_llm_start – LLM start with node metadata capture.
• on_retriever_start – Retriever start with node metadata capture.
• on_tool_start – Tool start with node metadata capture.

Attributes:

• emitted_events (frozenset[str]) –
• template_name (str) –
• template_package (str) –

Bind graph callbacks and optionally extract the graph's declared tools.

kyvvu_langchain.KyvvuLangGraphHandler.build_properties¶

build_properties(matched: Dict[str, Any], context: Dict[str, Any], step_run_id: str) -> Dict[str, Any]

Inject langgraph.node_name into the step properties.

Composes with the LangChain base hook (which forwards framework tags and a tool's metadata["kyvvu"] self-declaration) and then adds the LangGraph node name, so all enrichments are present before policy evaluation (and therefore visible to policy rules).

kyvvu_langchain.KyvvuLangGraphHandler.emit_task_start¶

emit_task_start(context: Dict[str, Any], task_id: Optional[str] = None, *, properties: Optional[Dict[str, Any]] = None) -> str

Open the task, adding properties.langgraph thread identifiers.

kyvvu_langchain.KyvvuLangGraphHandler.emitted_events¶

emitted_events: frozenset[str] = KyvvuLangChainHandler.emitted_events | {'on_interrupt'}

kyvvu_langchain.KyvvuLangGraphHandler.on_chain_error¶

on_chain_error(error: BaseException, *, run_id: UUID, parent_run_id: Optional[UUID] = None, **kwargs: Any) -> None

Intercept GraphInterrupt and emit step.gate instead of task.error.

All other errors delegate to the parent.

kyvvu_langchain.KyvvuLangGraphHandler.on_chain_start¶

on_chain_start(serialized: Dict[str, Any], inputs: Dict[str, Any], *, run_id: UUID, parent_run_id: Optional[UUID] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

Chain start; a root run records its checkpointer thread on task.start.

Raises:

• KyvvuBlockedError – If the task.start is blocked by policy.

kyvvu_langchain.KyvvuLangGraphHandler.on_chat_model_start¶

on_chat_model_start(serialized: Dict[str, Any], messages: List[List[Any]], *, run_id: UUID, parent_run_id: Optional[UUID] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

Chat model start with node metadata capture.

Raises:

• KyvvuBlockedError – If the policy engine blocks this model call.

kyvvu_langchain.KyvvuLangGraphHandler.on_llm_start¶

on_llm_start(serialized: Dict[str, Any], prompts: List[str], *, run_id: UUID, parent_run_id: Optional[UUID] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

LLM start with node metadata capture.

Raises:

• KyvvuBlockedError – If the policy engine blocks this LLM call.

kyvvu_langchain.KyvvuLangGraphHandler.on_retriever_start¶

on_retriever_start(serialized: Dict[str, Any], query: str, *, run_id: UUID, parent_run_id: Optional[UUID] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

Retriever start with node metadata capture.

Raises:

• KyvvuBlockedError – If the policy engine blocks this retrieval.

kyvvu_langchain.KyvvuLangGraphHandler.on_tool_start¶

on_tool_start(serialized: Dict[str, Any], input_str: str, *, run_id: UUID, parent_run_id: Optional[UUID] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None

Tool start with node metadata capture.

Raises:

• KyvvuBlockedError – If the policy engine blocks this tool call.

kyvvu_langchain.KyvvuLangGraphHandler.template_name¶

template_name: str = 'langgraph'

kyvvu_langchain.KyvvuLangGraphHandler.template_package¶

template_package: str = 'kyvvu_langchain.templates'

kyvvu_langchain.__version__¶

LangChain integration release version.