LangGraph

HighflameMiddleware for LangGraph — hooks on before_model, after_model, and wrap_tool_call. Available for Python and TypeScript. Session ID resolved from thread_id automatically.

HighflameMiddleware integrates Highflame Shield into LangGraph agents without modifying individual nodes. It implements the LangGraph AgentMiddleware interface and hooks into every model call and tool execution in the agent loop.

Available for Python and TypeScript.

Installation

Python

pip install 'highflame[langgraph]'

TypeScript

Basic Usage

Python

from highflame import Highflame
from highflame.integrations.langgraph import HighflameMiddleware

client = Highflame(api_key="hf_sk_...")
middleware = HighflameMiddleware(client, mode="enforce")

app = graph.compile(middleware=[middleware])

TypeScript

On a policy violation, HighflameMiddleware raises BlockedError before the blocked operation completes.

Constructor

Python

HighflameMiddleware(
    client: Highflame,
    *,
    mode: str = "enforce",
    session_id_key: str | None = None,
)

Parameter	Type	Default	Description
client	Highflame	required	Initialized Highflame client
mode	str	"enforce"	Enforcement mode: "enforce", "monitor", or "alert"
session_id_key	str | None	None	Key to read a session ID from LangGraph state. Falls back to thread_id automatically.

TypeScript

Session ID Resolution

The middleware resolves the session ID in this order:

1. state[session_id_key] (Python) / state[sessionIdKey] (TypeScript) — if set and the key is present in graph state

2. LangGraph thread_id — read from var_child_runnable_config / AsyncLocalStorage (requires langchain_core >= 0.3)

3. runtime.thread_id attribute — backward-compatibility fallback

4. None / undefined — no session tracking

Using LangGraph's thread_id as the session ID is the recommended pattern for multi-turn agents. It requires no additional configuration when thread_id is passed in the run config.

Python

result = await app.ainvoke(
    {"messages": [HumanMessage(content=user_input)]},
    config={"configurable": {"thread_id": "user-123-session-abc"}},
)

Or store the session ID explicitly in graph state:

middleware = HighflameMiddleware(client, session_id_key="session_id")
result = await app.ainvoke({"messages": [...], "session_id": "my-session"})

TypeScript

Hooks

HighflameMiddleware registers four hooks in the LangGraph middleware protocol:

before_model / beforeModel

Runs before the LLM is called. Extracts the last human message from state["messages"] and evaluates it as a prompt.

• Content type: "prompt", action: "process_prompt"

• On deny: raises BlockedError before the model is invoked

after_model / afterModel

Runs after the LLM responds. Extracts the last AI message from state["messages"] and evaluates it as a model response.

• Content type: "response", action: "process_prompt"

• On deny: raises BlockedError before the response is returned

wrap_tool_call / wrapToolCall

Wraps each tool execution. Evaluates the tool call before execution and the tool result after execution. The tool is never called if the pre-execution check is denied.

• Pre-execution: content type "tool_call", action "call_tool"

• Post-execution: content type "response", action "call_tool"

• On pre-execution deny: raises BlockedError, tool is not called

• On post-execution deny: raises BlockedError, result is not returned

wrap_model_call / wrapModelCall

Pass-through. Model calls are not individually guarded at this layer (prompts and responses are covered by before_model and after_model).

Enforcement Modes

Python

middleware = HighflameMiddleware(client, mode="enforce")  # block violations (default)
middleware = HighflameMiddleware(client, mode="monitor")  # observe without blocking
middleware = HighflameMiddleware(client, mode="alert")    # allow but trigger alerting

TypeScript

In monitor and alert mode, BlockedError is never raised. The agent proceeds normally regardless of policy decisions.

Complete Example

Python

from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

from highflame import Highflame, BlockedError
from highflame.integrations.langgraph import HighflameMiddleware

client = Highflame(api_key="hf_sk_...")
middleware = HighflameMiddleware(client, mode="enforce")

model = ChatAnthropic(model="claude-sonnet-4-6")

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode([shell_tool, web_tool]))
graph.set_entry_point("agent")

app = graph.compile(middleware=[middleware])

try:
    result = await app.ainvoke(
        {"messages": [HumanMessage(content=user_input)]},
        config={"configurable": {"thread_id": session_id}},
    )
except BlockedError as e:
    return {"error": f"Request blocked: {e.response.policy_reason}"}

TypeScript

Error Handling

Python

from highflame import BlockedError

try:
    result = await app.ainvoke({"messages": [HumanMessage(content=user_input)]})
except BlockedError as e:
    print(f"Blocked: {e.response.policy_reason}")
    print(f"Decision: {e.response.decision}")
    print(f"Signals: {e.response.signals}")

TypeScript

Requirements

Python

Package	Minimum Version
langchain	1.2+
langchain_core	1.2+
langgraph	1.2+
highflame	latest

TypeScript