LangChain Integration Python

Add Heed as a custom Tool in your LangChain agent. Before the agent executes a consequential action (like sending an email), Heed intercepts it and waits for human approval.

Prerequisites

Quick Start

pip install langchain langchain-openai httpx

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

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
  }'

Complete Example

This agent can send emails, but every send_email action is gated by Heed. The agent calls the Heed tool, which blocks until a human approves or rejects.

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

Before the agent sends an email, it calls the Heed tool.
If a policy requires approval, the tool blocks (polls) until
a human decides. If approved, the agent proceeds. If rejected,
it stops.

Requires: pip install langchain langchain-openai httpx
"""

import os
import time
import httpx
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

HEED_BASE = "https://heed.run/api/v1"
HEED_API_KEY = os.environ["HEED_API_KEY"]
POLL_INTERVAL = 3  # seconds between polls
MAX_POLL_WAIT = 300  # 5 minutes max

headers = {
    "Content-Type": "application/json",
    "X-API-Key": HEED_API_KEY,
}


@tool
def request_approval(action: str, summary: str, context: str) -> str:
    """Request human approval before executing a consequential action.
    Blocks until a decision is made (approved/rejected) or times out.

    Args:
        action: The action being gated, e.g. "send_email"
        summary: Human-readable description of what will happen
        context: JSON string with relevant details (amounts, recipients, etc.)

    Returns:
        "approved" or "rejected"
    """
    import json

    payload = {
        "agent_id": "langchain-email-agent",
        "action": action,
        "summary": summary,
        "context": json.loads(context) if context.startswith("{") else {"detail": context},
        "on_expire": "reject",
    }

    resp = httpx.post(f"{HEED_BASE}/hitl/requests", json=payload, headers=headers)
    resp.raise_for_status()
    data = resp.json()

    # Auto-approved (no matching policy)
    if data.get("status") == "approved":
        return "approved"

    # Pending — poll until decided
    poll_url = data.get("poll_url")
    if not poll_url:
        return "approved"

    elapsed = 0
    while elapsed < MAX_POLL_WAIT:
        time.sleep(POLL_INTERVAL)
        elapsed += POLL_INTERVAL

        poll_resp = httpx.get(poll_url, headers=headers)
        poll_resp.raise_for_status()
        status = poll_resp.json().get("status")

        if status in ("approved", "rejected"):
            return status

    return "rejected"  # timeout fallback


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email. Must be approved via Heed first."""
    # In a real app this would call your email API
    return f"Email sent to {to}: [{subject}] {body}"


# Build the agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an email assistant. When you need to send an email,
you MUST first call request_approval with action="send_email", a summary of
what the email contains, and context as a JSON string with keys "to", "subject".

Only proceed with send_email if request_approval returns "approved".
If it returns "rejected", tell the user the action was not approved."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

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

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

How It Works

1. The agent decides it needs to send an email and calls request_approval with action="send_email"
2. Heed checks the gate-emails policy → the action matches a rule that requires approval
3. Heed returns status: "pending" with a poll_url
4. The tool polls poll_url every 3 seconds until a human approves or rejects
5. The tool returns "approved" or "rejected"
6. The agent proceeds with send_email only if approved

Creating the Policy

The policy is what gates the send_email action. Without it, every request auto-approves. Create it with one curl call:

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

You can also add conditions — e.g., only gate emails to external domains or with more than 10 recipients:

"rules": [
  {
    "action": "send_email",
    "conditions": { "recipient_count": { ">": 10 } },
    "requires_approval": true
  }
]

Async Pattern

For async LangChain agents, use the async version:

import asyncio
import httpx

@tool
async def request_approval(action: str, summary: str, context: str) -> str:
    """Request human approval (async)."""
    import json

    payload = {
        "agent_id": "langchain-email-agent",
        "action": action,
        "summary": summary,
        "context": json.loads(context) if context.startswith("{") else {"detail": context},
        "on_expire": "reject",
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{HEED_BASE}/hitl/requests", json=payload, headers=headers)
        resp.raise_for_status()
        data = resp.json()

        if data.get("status") == "approved":
            return "approved"

        poll_url = data.get("poll_url")
        if not poll_url:
            return "approved"

        for _ in range(100):  # 5 min with 3s interval
            await asyncio.sleep(3)
            poll_resp = await client.get(poll_url, headers=headers)
            poll_resp.raise_for_status()
            status = poll_resp.json().get("status")
            if status in ("approved", "rejected"):
                return status

    return "rejected"

Using Callbacks Instead of Polling

Instead of blocking your agent on polling, provide a callback_url and Heed will POST the decision to your server when a human decides:

payload = {
    "agent_id": "langchain-email-agent",
    "action": "send_email",
    "summary": "Send Q3 report to alice@example.com",
    "context": {"to": "alice@example.com", "subject": "Q3 Report"},
    "on_expire": "reject",
    "callback_url": "https://your-server.com/heed/callback"
}

See the main docs for webhook payload format and HMAC signature verification.

Next Steps

• Full API documentation — all endpoints, authentication, webhook signing
• Swagger UI — interactive API explorer
• More integration guides — Claude Code / MCP, OpenAI Agents SDK
• Four Modes of Oversight — from strict gating to circuit breakers