Agent CertAgent Cert
Integration · LangChain

LangChain.

Drop Agent Cert into your LangChain agent as a custom tool, or as a callback handler that auto-certifies every AgentFinish. Two patterns below — pick based on whether you want the LLM to call the tool itself or to certify behind the scenes.

Pattern A — Custom tool the LLM can call

The LLM explicitly decides to certify. Best when the agent should produce both an answer AND a verify URL the user sees.

from langchain.tools import tool
import os, requests

API_BASE = os.environ["AGENT_CERT_API_BASE"]
API_KEY = os.environ["AGENT_CERT_API_KEY"]

@tool
def certify_decision(
    agent_id: str,
    user_intent: str,
    agent_output: str,
    tools_used: list[str],
    retrieval_summary: str = "",
) -> dict:
    """Submit a decision to Agent Cert for evaluation.
    Returns a signed, blockchain-anchored receipt with a public verify URL."""
    r = requests.post(
        f"{API_BASE}/v1/receipts",
        headers={"X-API-Key": API_KEY},
        json={
            "agent_id": agent_id,
            "platform": "langchain",
            "user_intent": user_intent,
            "agent_output": agent_output,
            "tools_used": tools_used,
            "retrieval_summary": retrieval_summary,
        },
    )
    r.raise_for_status()
    return r.json()

Then pass it to your agent:

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-7-sonnet")
agent = create_tool_calling_agent(llm, [certify_decision, ...other tools], prompt)
executor = AgentExecutor(agent=agent, tools=[certify_decision, ...other tools])

Pattern B — Callback handler (auto-certify every run)

The runtime auto-certifies every final answer with no LLM involvement. Best when you want certification as middleware, not a tool the model can skip.

from langchain_core.callbacks import BaseCallbackHandler
import os, requests

API_BASE = os.environ["AGENT_CERT_API_BASE"]
API_KEY = os.environ["AGENT_CERT_API_KEY"]

class AgentCertCallback(BaseCallbackHandler):
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self._tools_used: list[str] = []

    def on_tool_start(self, serialized, input_str, **kw):
        self._tools_used.append(serialized.get("name", "unknown"))

    def on_agent_finish(self, finish, **kw):
        # AgentFinish.return_values["output"] is the final answer
        # AgentFinish.log holds the chain of thought
        receipt = requests.post(
            f"{API_BASE}/v1/receipts",
            headers={"X-API-Key": API_KEY},
            json={
                "agent_id": self.agent_id,
                "platform": "langchain",
                "user_intent": kw.get("inputs", {}).get("input", ""),
                "agent_output": finish.return_values.get("output", ""),
                "tools_used": self._tools_used,
            },
        ).json()
        # Attach verify URL to the finish object
        finish.return_values["verify_url"] = receipt["verify_url"]
        self._tools_used.clear()

Then run your agent with the callback:

executor.invoke(
    {"input": "Approve this invoice if under $5,000"},
    config={"callbacks": [AgentCertCallback(agent_id="lc_invoice_001")]},
)

Async LCEL chains

For LangChain Expression Language (LCEL) chains, wrap the tail of the chain in a RunnableLambda that calls the REST API asynchronously with httpx.AsyncClient. The pattern is the same — just async / await.

Policy binding

Pre-create a Policy at /policies and bind it to your agent_id at /agents. Then you don't need to pass policy_rules per call — the network applies the bound policy automatically.

See also