LangChain Integration

LangChain + Skalor

Give any LangChain agent the ability to breach HTTP 402 paywalls with enterprise-grade fiduciary controls. Drop in one tool and every paid API call goes through the 5-gate firewall.

How it works

1

Agent hits 402

A vendor API returns HTTP 402 Payment Required

2

Tool clears gates

Skalor enforces 5 fiduciary gates and signs a payment credential

3

Agent retries

The agent attaches clearance headers and gets the data

Quickstart

1

Install dependencies

pip install skalor-sdk langchain langchain-openai

Requires langchain-core for the BaseTool interface. The tool works with any LangChain-compatible LLM.

2

Set environment variables

.env
export SKALOR_AGENT_ID="your-agent-uuid"
export SKALOR_API_KEY="zc_your_key_here"
export OPENAI_API_KEY="sk-..."

Get your Agent ID and API key from the Dashboard or the self-registration API.

3

Build a LangChain agent with Skalor

agent.py
from skalor.skalor_tool import SkalorPaymentTool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 1. Create the Skalor fiduciary tool
skalor_tool = SkalorPaymentTool()

# 2. Define a prompt that teaches the agent about 402 paywalls
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a research agent with a spending budget.
When you encounter an HTTP 402 Payment Required response,
use the request_fiduciary_clearance tool to get payment
authorization. Pass the full 402 response body as the
http_402_challenge parameter. After approval, retry the
request with the clearance_headers attached."""),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 3. Build the agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_openai_functions_agent(llm, [skalor_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[skalor_tool], verbose=True)

# 4. Run it
result = executor.invoke({
    "input": "Fetch the latest market analysis from api.datavendor.com"
})
4

Run it

When the agent encounters a paid API, it automatically invokes the request_fiduciary_clearance tool. All 5 gates are enforced transparently.

Agent: Calling request_fiduciary_clearance with vendor=api.datavendor.com, amount=$0.05
Skalor: Gate 1 PASS | Gate 2 PASS | Gate 3 PASS | Gate 4 PASS | Gate 5 PASS
Agent: Payment cleared. Retrying with clearance headers...
Agent: Got response from vendor. Here is the market analysis...

Configuration

Explicit credentials

Pass credentials directly instead of using environment variables:

explicit_config.py
from skalor.skalor_tool import SkalorPaymentTool

# Pass credentials explicitly instead of using env vars
tool = SkalorPaymentTool(
    agent_id="550e8400-e29b-41d4-a716-446655440000",
    api_key="zc_live_abc123def456...",
)

Tool Parameters

When the LLM invokes the tool, it passes these parameters automatically based on the 402 response:

ParameterTypeDescription
merchant_urlstringFull URL of the vendor that returned 402
requested_amount_usdfloatDollar amount the vendor is charging
protocolstringPayment protocol: "mpp" (default)
http_402_challengestringComplete 402 response body as JSON string
reasoningstringAudit trail reason (min 15 chars)

Examples

Multi-tool agent

Combine Skalor with free tools so the agent only pays when it needs to:

multi_tool_agent.py
from skalor.skalor_tool import SkalorPaymentTool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Combine Skalor with other tools
tools = [
    SkalorPaymentTool(),      # Fiduciary clearance for paid APIs
    DuckDuckGoSearchRun(),     # Free web search
]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a research agent. Use free search first.
When a paid API returns HTTP 402, use request_fiduciary_clearance
to get payment authorization before retrying."""),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Error handling

When a fiduciary gate blocks a payment, the tool raises SkalorFirewallDenied:

error_handling.py
from skalor.skalor_tool import SkalorPaymentTool, SkalorFirewallDenied

tool = SkalorPaymentTool()

try:
    result = tool._run(
        merchant_url="https://api.openai.com/v1/chat/completions",
        requested_amount_usd=500.00,
        protocol="mpp",
        http_402_challenge='{"amount": 500, "currency": "USD"}',
        reasoning="Bulk inference run for quarterly report generation",
    )
    print("Approved:", result)
except SkalorFirewallDenied as e:
    print(f"Blocked by gate: {e.error_code}")
    print(f"Reason: {e.detail}")
    # e.error_code is machine-readable:
    # EXCEEDS_DAILY_LIMIT, VENDOR_NOT_AUTHORIZED, etc.

Async execution

The tool supports async for LCEL chains and concurrent execution:

async_agent.py
import asyncio
from skalor.skalor_tool import SkalorPaymentTool

tool = SkalorPaymentTool()

async def main():
    # Async version for LCEL chains and concurrent tool execution
    result = await tool._arun(
        merchant_url="https://api.openai.com/v1/chat/completions",
        requested_amount_usd=0.05,
        protocol="mpp",
        http_402_challenge='{"amount": 0.05, "currency": "USD"}',
        reasoning="Async inference call for pipeline stage 3",
    )
    print(result)

asyncio.run(main())

5 Fiduciary Gates

Every payment goes through 5 gates before any money moves. The LLM never handles funds directly.

  1. Kill Switch — Is the agent active?
  2. Vendor Allowlist — Is this vendor authorized?
  3. Per-Tx Limit — Within single-payment policy?
  4. Daily Budget — Under the daily cap?
  5. Human-in-the-Loop — Needs CFO sign-off?
Learn more about gates