LangGraph Enrichment — RichAPI as a Tool Node

Pick this when you want the model to see three tools instead of the whole catalog, and when you want to normalise the response before it reaches the context window.

import os, httpx
from langchain_core.tools import tool

BASE = "https://api.richapi.ai/api/v1"
HEADERS = {"x-api-key": os.environ["RICHAPI_KEY"]}

def _post(endpoint: str, payload: dict) -> dict:
 r = httpx.post(f"{BASE}/{endpoint}", json=payload, headers=HEADERS, timeout=None)
 r.raise_for_status() # non-2xx never bills; let it raise
 return r.json()

@tool
def find_work_email(first_name: str, last_name: str, company_domain: str) -> dict:
 """Find a verified work email. Returns success, the email when found,
 billed, and execution_log listing each source attempt.
 A miss returns success=False and billed=False and costs nothing."""
 return _post("email_finder", {
 "first_name": first_name, "last_name": last_name, "company_domain": company_domain
 })

@tool
def enrich_company_by_linkedin(linkedin_url: str) -> dict:
 """Company data from a LinkedIn company URL: size, industry, specialties.
 Takes a LinkedIn URL, NOT a domain. Bills on any 2xx, empty result included."""
 return _post("enrich_company", {"linkedin_url": linkedin_url})

The docstrings are the integration. A model that does not know enrich_company wants a LinkedIn URL will confidently hand it a domain, and you will pay for the 2xx that comes back thin. Say it in the docstring, in capitals, and say what a miss costs — a model that knows retries are free on email_finder and expensive on enrich_profile makes better decisions than one guessing.

Leave timeout=None. Enrichment waterfalls take as long as the sources take, and an aggressive client timeout turns a call you were not charged for into a call you cannot see the result of.

Standard prebuilt loop, nothing exotic:

from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

tools = [find_work_email, enrich_company_by_linkedin]
model = ChatAnthropic(model="claude-sonnet-4-5").bind_tools(tools)

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

g = StateGraph(MessagesState)
g.add_node("model", call_model)
g.add_node("tools", ToolNode(tools))
g.add_edge(START, "model")
g.add_conditional_edges("model", tools_condition)
g.add_edge("tools", "model")
graph = g.compile()

graph.invoke(
    {"messages": [("user", "Find a work email for Jane Doe at acme.com.")]},
    {"recursion_limit": 12},
)

Set recursion_limit. An agent that discovers it can call an enrichment endpoint in a loop will do exactly that, and the endpoints that bill on every 2xx do not care that it was an accident.

Carry a credits_spent int in your state schema, add to it on each tool return, and put a conditional edge in front of phone_finder that hits an interrupt() above a threshold. Human-in-the-loop on the 25-credit endpoint is cheap insurance; on a 1-credit lookup it is theatre. Rates are on pricing.

Data layer only — no prospecting UI, no sequencer, no CRM app. For the same tools in a chat client instead of a graph, see Claude or Cursor, with the rest of the family under integrations. For the non-agentic version of the same pipeline, agent-driven account research and list build and verify.

100 free credits, no card. Paste the find_work_email tool above into an existing graph and run one query — it is a five-minute test and the miss costs nothing. Endpoint reference: email finder, person enrichment, full list under platform.