Custom frameworks

Wrap any agent framework with ArmorIQ using the raw-client pattern. Concrete wiring for CrewAI, LangChain, OpenAI Assistants, and Anthropic tool-use — plus a template for anything else.

Any framework that lets you intercept tool calls can be wrapped with ArmorIQ in about ten lines. This page shows the universal pattern, then concrete recipes for four popular frameworks. If yours isn't listed, the pattern at the top works for anything.

The universal pattern

Every tool call in any framework boils down to "run this action with these params." The ArmorIQ three-step flow sits between intent and execution:

from armoriq_sdk import ArmorIQClient

client = ArmorIQClient()  # one per process


def guarded_tool_call(
    mcp: str,
    action: str,
    params: dict,
    user_email: str,
    llm: str = "gpt-4o",
):
    """Drop-in replacement for any direct tool call."""
    plan = client.capture_plan(
        llm=llm,
        prompt=f"Call {mcp}.{action}",
        plan={
            "goal": action,
            "steps": [{"action": action, "mcp": mcp, "params": params}],
        },
    )
    token = client.get_intent_token(plan, validity_seconds=300)
    return client.invoke(
        mcp=mcp,
        action=action,
        intent_token=token,
        params=params,
        user_email=user_email,
    )

That function is all you need. Call it from inside your framework's tool-call path and every invocation goes through ArmorIQ's plan → token → invoke loop — signed, policy-checked, audited.

The examples below all use the pattern above. The only thing that changes is where you call it — that's the framework-specific part.

CrewAI

CrewAI tools subclass BaseTool. Override _run() to route the call through guarded_tool_call:

from crewai.tools import BaseTool
from armoriq_sdk import ArmorIQClient

client = ArmorIQClient()


class BookFlightTool(BaseTool):
    name: str = "book_flight"
    description: str = "Book a flight with given origin, destination, and date."

    def _run(self, origin: str, destination: str, date: str, user_email: str):
        plan = client.capture_plan(
            llm="gpt-4o",
            prompt=f"Book flight from {origin} to {destination}",
            plan={
                "goal": "book_flight",
                "steps": [
                    {
                        "action": "book_flight",
                        "mcp": "travel-mcp",
                        "params": {"from": origin, "to": destination, "date": date},
                    },
                ],
            },
        )
        token = client.get_intent_token(plan)
        result = client.invoke(
            mcp="travel-mcp",
            action="book_flight",
            intent_token=token,
            params={"from": origin, "to": destination, "date": date},
            user_email=user_email,
        )
        return result.data


# Use as a regular CrewAI tool
from crewai import Agent, Task, Crew

researcher = Agent(
    role="Travel booker",
    goal="Book flights for users",
    tools=[BookFlightTool()],
)
task = Task(description="Book SFO→JFK on 2026-05-10", agent=researcher)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff(inputs={"user_email": "alice@acme.com"})

Generic CrewAI adapter

If you want a single adapter that wraps every CrewAI tool:

class ArmorIQTool(BaseTool):
    """CrewAI tool that routes every call through ArmorIQ."""

    name: str
    description: str
    mcp: str          # which MCP hosts this tool
    action: str       # tool name on the MCP
    user_email: str   # the end user scope

    def _run(self, **params):
        return guarded_tool_call(
            mcp=self.mcp,
            action=self.action,
            params=params,
            user_email=self.user_email,
        ).data

One class, many instances — one per tool you want to expose.

LangChain

LangChain's BaseTool has the same shape. Override _run():

from langchain.tools import BaseTool
from armoriq_sdk import ArmorIQClient

client = ArmorIQClient()


class BookFlightTool(BaseTool):
    name = "book_flight"
    description = "Book a flight"

    def _run(self, origin: str, destination: str, date: str, user_email: str = None):
        plan = client.capture_plan(
            llm="gpt-4o",
            prompt=f"Book flight from {origin} to {destination}",
            plan={
                "goal": "book_flight",
                "steps": [
                    {
                        "action": "book_flight",
                        "mcp": "travel-mcp",
                        "params": {"from": origin, "to": destination, "date": date},
                    },
                ],
            },
        )
        token = client.get_intent_token(plan)
        return client.invoke(
            mcp="travel-mcp",
            action="book_flight",
            intent_token=token,
            params={"from": origin, "to": destination, "date": date},
            user_email=user_email,
        ).data

Pass user_email via the agent executor's inputs:

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(agent=agent, tools=[BookFlightTool()])
agent_executor.invoke({"input": "Book SFO→JFK", "user_email": "alice@acme.com"})

Async variant

For _arun, same idea — ArmorIQClient is sync, so either call it in a thread executor or use an async HTTP client wrapper.

OpenAI Assistants

The OpenAI Assistants API runs tools in a structured loop with run.required_action.submit_tool_outputs.tool_calls. Intercept each tool_call before submitting outputs:

from openai import OpenAI
from armoriq_sdk import ArmorIQClient

openai_client = OpenAI()
armoriq_client = ArmorIQClient()


def run_with_armoriq(assistant_id: str, thread_id: str, user_email: str):
    run = openai_client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant_id,
    )

    while True:
        run = openai_client.beta.threads.runs.retrieve(
            thread_id=thread_id, run_id=run.id
        )

        if run.status == "completed":
            return

        if run.status == "requires_action":
            tool_outputs = []
            for call in run.required_action.submit_tool_outputs.tool_calls:
                # Each tool_call has: id, function.name, function.arguments
                import json

                args = json.loads(call.function.arguments)
                mcp, action = call.function.name.split(".", 1)  # e.g. "travel-mcp.book_flight"

                plan = armoriq_client.capture_plan(
                    llm="gpt-4o",
                    prompt=f"Call {call.function.name}",
                    plan={
                        "goal": action,
                        "steps": [{"action": action, "mcp": mcp, "params": args}],
                    },
                )
                token = armoriq_client.get_intent_token(plan)
                result = armoriq_client.invoke(
                    mcp=mcp,
                    action=action,
                    intent_token=token,
                    params=args,
                    user_email=user_email,
                )

                tool_outputs.append({
                    "tool_call_id": call.id,
                    "output": json.dumps(result.data),
                })

            openai_client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run.id,
                tool_outputs=tool_outputs,
            )

        elif run.status in ("failed", "cancelled", "expired"):
            raise RuntimeError(f"Run ended: {run.status}")

Name your tools with a dotted prefix (travel-mcp.book_flight) so you can parse (mcp, action) out of call.function.name. Alternatively, maintain a tool registry mapping tool name → MCP.

Anthropic tool-use

Anthropic's Messages API returns tool_use blocks in the response. Intercept each block before executing:

from anthropic import Anthropic
from armoriq_sdk import ArmorIQClient

claude = Anthropic()
armoriq_client = ArmorIQClient()

# Your tool registry — maps Claude tool name → (mcp, action)
TOOL_REGISTRY = {
    "book_flight": ("travel-mcp", "book_flight"),
    "search_hotels": ("travel-mcp", "search_hotels"),
}


def chat_with_armoriq(messages, tools, user_email: str):
    while True:
        response = claude.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason != "tool_use":
            return response

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue

            mcp, action = TOOL_REGISTRY[block.name]
            plan = armoriq_client.capture_plan(
                llm="claude-sonnet-4-5",
                prompt=f"Call {block.name}",
                plan={
                    "goal": action,
                    "steps": [{"action": action, "mcp": mcp, "params": block.input}],
                },
            )
            token = armoriq_client.get_intent_token(plan)
            result = armoriq_client.invoke(
                mcp=mcp,
                action=action,
                intent_token=token,
                params=block.input,
                user_email=user_email,
            )

            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result.data),
            })

        # Feed results back for the next turn
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

Your own framework

Any framework with a "here's the tool about to run" hook works. The pattern:

Intercept — find the hook where your framework calls a tool.
Plan — call client.capture_plan(llm=..., prompt=..., plan={...}) with a steps entry for the tool call.
Token — client.get_intent_token(plan).
Invoke — client.invoke(mcp, action, token, params, user_email=...).
Return — pass result.data back to your framework as the tool output.

If your framework has no tool-call hook at all (raw LLM with no structured tools), you can still use the SDK directly — build the plan manually from the LLM's output, get a token, invoke through ArmorIQ's proxy.

Error handling across frameworks

Regardless of framework, the same ArmorIQ exceptions fire and should be caught near your guarded_tool_call invocation:

from armoriq_sdk import (
    PolicyBlockedException,
    PolicyHoldException,
    TokenExpiredException,
    IntentMismatchException,
)

try:
    result = guarded_tool_call(...)
except PolicyBlockedException as e:
    # Tool not allowed for this user — surface a "not authorized" response
    return {"error": "not authorized", "reason": str(e)}
except PolicyHoldException as e:
    # Needs approval — create a delegation, tell the user "pending"
    return {"error": "pending approval", "delegation_id": e.delegation_id}
except TokenExpiredException:
    # Transient — retry with a fresh token
    ...

See Python SDK — exceptions for the full list.

Performance tips

One ArmorIQClient per process. It's thread-safe and reuses HTTP connections.
capture_plan is local. It computes hashes; no network.
get_intent_token is network. Cache tokens per-plan if you'll invoke the same plan multiple times within validity_seconds.
invoke is network. One round-trip per tool call.
client.resolve_user(email) caches for 5 minutes — safe to call at the start of every request.

The universal pattern

CrewAI

Generic CrewAI adapter

LangChain

Async variant

OpenAI Assistants

Anthropic tool-use

Your own framework

Error handling across frameworks

Performance tips

Next steps

Python SDK

Google ADK

Policies

On this page