LangChain Integration Python

Add Heed human-in-the-loop approval to any LangChain tool with a single decorator. Before your agent executes a consequential action, Heed intercepts it and waits for human approval — no manual polling code required.

Prerequisites

RequirementDetails
Heed API KeyGet one free at heed.run/signup — looks like hd_live_... or hd_test_...
Python3.10 or later
LangChainlangchain-core ≥ 0.3 and langchain-openai
Heed SDKpip install heed[langchain]

Use hd_test_... keys during development — they work identically but are flagged in the audit log.

Quick Start

1

Install dependencies

pip install "heed[langchain]" langchain langchain-openai
2

Set environment variables

export HEED_API_KEY="hd_live_your_key_here"
export OPENAI_API_KEY="sk-your-openai-key"
3

Create a policy that gates send_email

curl -X POST https://heed.run/api/v1/hitl/policies \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $HEED_API_KEY" \
  -d '{
    "name": "gate-emails",
    "description": "Require approval before sending emails",
    "rules": [
      {
        "action": "send_email",
        "requires_approval": true
      }
    ],
    "is_active": true
  }'
4

Add the Heed tool to your agent

Copy the complete example below into a file and run it.

Complete Example

This agent can send emails, but every send_email action is gated by Heed. The @heed_gate decorator wraps your LangChain tool and blocks execution until a human approves or rejects. No manual polling code needed.

"""
LangChain + Heed: Gate send_email behind human approval.

The @heed_gate decorator wraps any LangChain @tool. When the agent
calls the tool, Heed checks your policy. If approval is required,
execution blocks until a human decides. If auto-approved by policy,
the tool runs immediately.

Requires: pip install "heed[langchain]" langchain langchain-openai
"""

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from heed.langchain import heed_gate, inject_heed_config

# Step 1: Wrap your tool with @heed_gate
@heed_gate(
    action="send_email",
    mode="hitl_gate",             # or "grace_period" for auto-approve after timeout
    summary_template="Send email to {to}: {subject}",
    context_fields=["to", "subject"],
)
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    # This only executes AFTER human approval
    return f"Email sent to {to}: [{subject}] {body}"

# Step 2: Inject Heed credentials into your agent config
config = inject_heed_config({}, api_key=os.environ["HEED_API_KEY"])

# Step 3: Build your agent normally
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an email assistant. Use send_email to send emails."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [send_email], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[send_email], verbose=True)

# Step 4: Run with Heed config
result = agent_executor.invoke(
    {"input": "Send an email to alice@example.com about the Q3 report"},
    config=config,
)
print(result["output"])

No policy yet? Tools run without gating when no matching policy exists. Create a policy (step 3 above) to start gating specific actions.

How It Works

  1. You add @heed_gate above your @tool — that's the only code change
  2. When the agent calls the gated tool, GatedTool submits a HITL request to Heed
  3. Heed checks your policy → if the action requires approval, execution blocks until a human decides
  4. If auto-approved by policy, the tool runs immediately (zero latency)
  5. If approved by human, the tool executes normally
  6. If denied or expired, the tool raises HeedError — the agent knows to stop

The decorator works with both decorator orders: @heed_gate @tool or @tool then heed_gate()(my_tool).

Callback Handler (LangGraph)

For LangGraph or agents where you can't modify tool definitions, use HeedCallbackHandler to intercept tool calls at the callback level:

from heed.langchain import HeedCallbackHandler

handler = HeedCallbackHandler(
    api_key=os.environ["HEED_API_KEY"],
    gated_tools={
        "delete_record": {"action": "delete_record", "mode": "hitl_gate"},
        "send_email": {
            "action": "send_email",
            "mode": "grace_period",
            "grace_seconds": 30,
            "summary_template": "Send email to {to}: {subject}",
        },
    },
)

# Ungated tools pass through automatically
result = graph.invoke(
    {"messages": [HumanMessage("Delete record 42")]},
    config={"callbacks": [handler]},
)

Approval Modes

ModeBehaviorUse When
hitl_gateBlock until human approvesHigh-risk actions (payments, deletions, PII access)
grace_periodAuto-approve after N seconds unless human objectsMedium-risk actions with SLA requirements
hotl_monitorExecute immediately, log for reviewLow-risk actions needing audit trail
circuit_breakerAuto-approve until error threshold, then gateHigh-volume actions that need fail-safe

hotl_monitor and circuit_breaker modes are coming soon. hitl_gate and grace_period are live now.

API Reference

heed_gate decorator

ParameterTypeDefaultDescription
actionstrrequiredAction name for policy matching
modestr | NoneNoneOverride mode: hitl_gate or grace_period
grace_secondsint | NoneNoneAuto-approve timeout (grace_period only)
summary_templatestr | NoneNoneFormat string for approval summary, e.g. "Send to {to}"
context_fieldslist | NoneNoneTool kwargs to include in Heed context
agent_idstr | NoneNoneAgent identifier
auto_skipboolTrueIf no client found, execute without gating
poll_timeoutfloat300.0Max seconds to wait for decision
poll_intervalfloat2.0Seconds between polls

HeedCallbackHandler

ParameterTypeDescription
api_keystrHeed API key
base_urlstrHeed API base URL (default: https://heed.run)
gated_toolsdictTool name → gate config mapping
auto_approve_policyboolAllow ungated tools to pass through

inject_heed_config

Injects Heed credentials into a RunnableConfig so gated tools can find them:

config = inject_heed_config(
    {"configurable": {"thread_id": "user-1"}},
    api_key="hd_live_abc123",
)
# config["configurable"]["heed_api_key"] == "hd_live_abc123"

Next Steps