LangChain Integration

ArmorIQLangChain wires ArmorIQ into LangChain as a callback handler, so every tool call your agent makes goes through ArmorIQ's plan, token, policy, and audit pipeline. You keep your existing agent and tools. You just pass one extra callback when you invoke the agent.

Install

pip install "armoriq-sdk[langchain]"

npm install @armoriq/sdk
# plus @langchain/core and your agent packages

Minimal example

import os
from armoriq_sdk import ArmorIQClient
from armoriq_sdk.integrations.langchain import ArmorIQLangChain

# Once per process
armoriq = ArmorIQLangChain(
    armoriq_client=ArmorIQClient(api_key=os.environ["ARMORIQ_API_KEY"]),
    mode="sdk",
)

# Per request: for_user returns a LangChain callback handler
handler = armoriq.for_user("alice@example.com", goal=user_message)

agent_executor.invoke(
    {"input": user_message},
    config={"callbacks": [handler]},
)

import { ArmorIQClient } from '@armoriq/sdk';
import { ArmorIQLangChain } from '@armoriq/sdk/dist/integrations/langchain';

// Once per process
const armoriq = new ArmorIQLangChain({
  client: new ArmorIQClient({ apiKey: process.env.ARMORIQ_API_KEY! }),
  mode: 'sdk',
});

// Per request: forUser returns a LangChain callback handler
const handler = await armoriq.forUser('alice@example.com', { goal: userMessage });

await agentExecutor.invoke(
  { input: userMessage },
  { callbacks: [handler] },
);

That's the whole change. No wrapping your agent, no editing your tools, no building plans by hand. The handler wires two LangChain callbacks:

Enforcement is fail-closed: a tool runs only on an explicit allow. A block, a hold that is not approved in time, or any enforcement error stops the tool. The handler raises on a block or hold, which LangChain surfaces as a tool error and skips execution.

What you add to an existing agent

If you already have a LangChain agent, the diff is a few lines:

+ from armoriq_sdk import ArmorIQClient
+ from armoriq_sdk.integrations.langchain import ArmorIQLangChain
+ armoriq = ArmorIQLangChain(armoriq_client=ArmorIQClient(api_key=API_KEY), mode="sdk")

  def handle_chat(user_email: str, message: str):
+     handler = armoriq.for_user(user_email, goal=message)
      agent_executor.invoke(
          {"input": message},
+         config={"callbacks": [handler]},
      )

Live enforcement events

Pass an on_event callback (onEvent in TypeScript) to stream the enforcement lifecycle into your own UI. The kinds are hold, approved, rejected, timeout, block, and error.

def on_event(kind, payload):
    # kind in {"hold","approved","rejected","timeout","block","error"}
    stream_to_ui(kind, payload)

handler = armoriq.for_user("alice@example.com", goal=user_message, on_event=on_event)

const handler = await armoriq.forUser('alice@example.com', {
  goal: userMessage,
  onEvent: (kind, payload) => streamToUi(kind, payload),
});

On a hold, the handler emits hold, then waits asynchronously for the delegation decision. Approved lets the tool run; rejected or timeout stops it. The wait is non-blocking, so your app can stream held to approved to success live while it waits.

ArmorIQLangChain parameters

for_user(email, *, goal=None, on_event=None)

Returns a LangChain callback handler. Pass it via config={"callbacks": [handler]} (Python) or { callbacks: [handler] } (TypeScript) on the agent invocation. The handler caches plan state for the request, so tool calls that aren't in the plan get blocked before they hit the MCP.

Use in Langflow (no-code, zero config)

Prefer building visually instead of writing LangChain code? The Python SDK ships this same ArmorIQLangChain integration as a drag-in Langflow node, the ArmorIQ Tool Calling Agent, auto-discovered by Langflow with no config. See the dedicated Langflow integration page, Python-only, since Langflow itself is a Python application.

Reference

The integration source lives in the SDK repos: Python and TypeScript, under integrations/langchain.