Blog / AI agents

LangChain Custom Tool for Live Web Data: Python Tutorial

22 Jul 2026Perceptlayer11 min read

A LangChain custom tool for an API is a Python function with a typed argument schema and a docstring the model can read. This tutorial wraps two Perceptlayer endpoints, Google Search and Google News, in that shape. You get validated arguments, retries only for errors that are worth retrying, and a hard credit cap per agent run. The whole thing is one file of Python, and a successful call costs 1 credit.

What does a LangChain custom tool for an API need?

A tool needs four things: a name, a description, an input schema and a function that returns text. LangChain builds the schema the model sees from your type hints, and you can supply a Pydantic model for richer constraints and field descriptions, as the LangChain tools documentation describes. The model never sees your code, only that schema and the docstring.

For a web data API, that means three decisions. Which endpoint parameters to expose, which to hide, and what the function returns when something goes wrong. Exposing every parameter makes the schema noisy and the model sloppy. Exposing too few leaves the agent unable to narrow a search by language or country.

We'll expose the parameters an agent actually reasons about and leave the rest out. You can add more later from the endpoint pages, such as the Search Light parameters.

Which endpoints and parameters does the tool call?

The tool calls two endpoints, each costing 1 credit on success. GET /v1/google-search/search-light takes a required query and a set of optional filters. GET /v1/google-news/news has no required parameter at all, though in practice you'll pass q. Both are on the Google Search API and Google News API pages, where you can run them in the playground first.

Some constraints come straight from the catalogue. hl and gl accept up to 2 characters. safe is one of active or off. sort is one of date, date:r or date:d. query on Search Light is capped at 2048 characters, and q on News at 500. The schema below encodes exactly these, so a malformed call fails locally, before the network, and costs nothing.

Search Light is one of seven Google Search endpoints. If your agent needs more filters, Search Advanced takes many more optional parameters, such as tbs, tbm and device, and Search AI Overview is a separate endpoint.

How do you set up the project?

Setup is one install and one environment variable. You need Python 3.10 or later (the LangChain install docs list 3.10+ as the requirement), a model provider key for the agent in step 5, and a Perceptlayer key. Free accounts get 1,000 credits every month after verifying an email address, no credit card, so everything here fits in the free tier. Get a free API key and export it.

pip install langchain langchain-anthropic httpx pydantic
export API_KEY="sk_live_..."
export ANTHROPIC_API_KEY="..."

Keys start with sk_live_ and go in the X-API-Key header, which the client below sets once. The authentication docs cover the query-parameter alternative for places where you can't set headers. The LangChain packages come from the same install docs: langchain plus a provider package, here langchain-anthropic. We use httpx because it lets you set separate connect, read, write and pool timeouts, but requests works the same way.

Step 1: How do you define typed arguments?

Define one Pydantic model per tool, with a Field(description=...) on every attribute. The model reads those descriptions as instructions, so write them for it: say what the value is for, not just what type it is. Pydantic's Field accepts constraints such as max_length (Pydantic field docs), and a Literal type limits a value to a fixed set, so we encode the catalogue's limits in the schema instead of leaving them to the API.

from typing import Literal, Optional

from pydantic import BaseModel, Field


class SearchInput(BaseModel):
    query: str = Field(
        max_length=2048,
        description="The search query, as you would type it into Google.",
    )
    num: Optional[int] = Field(
        default=None, description="Number of results to request."
    )
    page: Optional[int] = Field(
        default=None, description="Results page to fetch, for follow-up queries."
    )
    hl: Optional[str] = Field(
        default=None, max_length=2, description="Interface language, e.g. 'en'."
    )
    gl: Optional[str] = Field(
        default=None, max_length=2, description="Country code for results, e.g. 'us'."
    )
    safe: Optional[Literal["active", "off"]] = Field(
        default=None, description="SafeSearch setting."
    )
    sort: Optional[Literal["date", "date:r", "date:d"]] = Field(
        default=None, description="Sort order. Leave unset for relevance."
    )


class NewsInput(BaseModel):
    q: Optional[str] = Field(
        default=None, max_length=500, description="News search query."
    )
    gl: Optional[str] = Field(
        default=None, max_length=2, description="Country code, e.g. 'us'."
    )
    hl: Optional[str] = Field(
        default=None, max_length=2, description="Language code, e.g. 'en'."
    )
    page: Optional[int] = Field(default=None, description="Results page to fetch.")

Every optional field defaults to None, and the client drops None values before sending. That keeps us from stating defaults the catalogue doesn't document: if the model doesn't set a parameter, the API applies its own behaviour.

Step 2: How do you build a budget that caps credits per run?

A credit budget is a counter with a ceiling. Before each call the client reserves the call's cost, and only if the reservation fits under the cap does the request go out. Afterwards it settles the reservation to the X-Credits-Used value from the response. Every successful response carries that header, plus X-Credits-Remaining for the account balance after the call.

Reserving matters because agents can issue several tool calls at once. A plain check-then-record counter would let every parallel call pass the check before any of them recorded a cost. A lock around the reservation closes that gap.

import threading


class CreditBudget:
    def __init__(self, cap: int):
        self.cap = cap
        self.spent = 0  # settled spend plus reservations still in flight
        self._lock = threading.Lock()

    def reserve(self, cost: int) -> bool:
        with self._lock:
            if self.spent + cost > self.cap:
                return False
            self.spent += cost
            return True

    def settle(self, reserved: int, used: int) -> None:
        with self._lock:
            self.spent += used - reserved

Why a cap in your own code when the account is prepaid anyway? Because prepaid protects the account, not the run. When the balance hits zero, calls return insufficient_credits (402) and nothing more is debited. LangGraph already bounds a run with a recursion limit that raises GraphRecursionError when exceeded, and you can lower it per call with config={"recursion_limit": 5}. That limit counts graph steps, not credits. The cap is a second limit, denominated in the thing you pay for, so "the agent got stuck" becomes a bounded, known cost. The cost-control article AI Agent API Cost Control goes further on budgets across runs.

The arithmetic is simple. A cap of 10 credits at 1 credit per call is at most 10 calls. A run at the ceiling is 10 calls × 1 credit = 10 credits, which is 10 of your 1,000 free monthly credits, or 10 of the 10,000 in a Starter pack. See pricing for what credits cost.

Step 3: Which errors should the tool retry?

Retry the three codes that describe a temporary condition and nothing else. rate_limited (429), source_error (502) and temporarily_unavailable (503) can clear on their own. invalid_api_key (401), insufficient_credits (402), restricted_endpoint (403), not_found (404) and invalid_parameters (422) are properties of your request or account. Resending the same call gets the same answer.

Errors arrive as an HTTP status and a JSON body shaped {"error": {"code", "message", "request_id"}}. The errors docs are the place to check how errors are reported. An HTTP error response returns X-Credits-Used: 0, so retrying one that fails again costs nothing. You're spending time, not credits. Keep the attempt count low so a stuck call doesn't stall the agent, and see the rate limit docs if you hit 429 often.

Two cases need more care. First, 429 and 503 responses carry a Retry-After header with the seconds to wait, so the loop uses it when present and falls back to exponential backoff when it's missing. Second, a call with no response at all is a different animal. Requests have no idempotency key, so one lost to a read timeout may have succeeded and been billed. The client therefore resends only when the connection itself failed, which means the request never arrived. For any other transport error it keeps the reserved credits counted against the cap and tells the model not to repeat the call.

import json
import os
import random
import time

import httpx

BASE_URL = "https://api.perceptlayer.com"
RETRYABLE_STATUS = {429, 502, 503}
STOP_CODES = {"invalid_api_key", "insufficient_credits", "restricted_endpoint"}
MAX_ATTEMPTS = 3
MAX_WAIT = 30  # seconds; if Retry-After asks for more, give up instead
MAX_CHARS = 6000  # our own limit, to protect the model's context window
# Fields from the recorded Search and News responses that a model can use.
KEEP = ("position", "title", "link", "snippet", "source", "iso_date")
RESULT_KEYS = ("organic_results", "news_results")


def project(text: str) -> str:
    """Keep a few fields per result, then drop whole results to fit MAX_CHARS."""
    try:
        data = json.loads(text)
    except ValueError:
        return text[:MAX_CHARS]
    items = next((data[k] for k in RESULT_KEYS if isinstance(data.get(k), list)), None)
    if items is None:  # a shape we have no recorded sample for
        return text[:MAX_CHARS]
    kept = []
    for item in items:
        row = {k: item[k] for k in KEEP if k in item}
        if isinstance(row.get("source"), dict):  # News nests the name
            row["source"] = row["source"].get("name")
        if len(json.dumps(kept + [row])) > MAX_CHARS:
            break
        kept.append(row)
    return json.dumps(kept)


class PerceptlayerClient:
    def __init__(self, budget: CreditBudget):
        self.budget = budget
        self.http = httpx.Client(
            base_url=BASE_URL,
            headers={"X-API-Key": os.environ["API_KEY"]},
            timeout=30,
        )

    def get(self, path: str, params: dict, cost: int = 1) -> str:
        if not self.budget.reserve(cost):
            return (
                f"Credit cap reached: {self.budget.spent} of {self.budget.cap} "
                "credits used this run. Do not call again; answer from the "
                "results you already have."
            )
        clean = {k: v for k, v in params.items() if v is not None}

        for attempt in range(1, MAX_ATTEMPTS + 1):
            try:
                resp = self.http.get(path, params=clean)
            except (httpx.ConnectError, httpx.ConnectTimeout) as exc:
                # No connection, so the request never arrived: safe to resend.
                if attempt < MAX_ATTEMPTS:
                    time.sleep(self._backoff(attempt))
                    continue
                self.budget.settle(cost, 0)
                return f"Could not connect after {attempt} attempts: {type(exc).__name__}."
            except httpx.TransportError as exc:
                # The reply was lost. The call may have succeeded and been
                # billed, so keep the reservation and do not resend.
                return (
                    f"No reply ({type(exc).__name__}). The call may have been "
                    "billed, so it counts against the cap. Do not repeat it."
                )

            if resp.is_success:
                used = int(resp.headers.get("X-Credits-Used", cost))
                self.budget.settle(cost, used)
                return project(resp.text)

            code, message = self._parse_error(resp)
            if resp.status_code in RETRYABLE_STATUS and attempt < MAX_ATTEMPTS:
                delay = self._delay(resp, attempt)
                if delay is not None:
                    time.sleep(delay)
                    continue
            self.budget.settle(cost, 0)  # an error response is not billed
            hint = " Do not retry." if code in STOP_CODES else ""
            return f"Request failed ({resp.status_code} {code}): {message}{hint}"

        self.budget.settle(cost, 0)
        return "Request failed."

    @staticmethod
    def _parse_error(resp: httpx.Response) -> tuple[str, str]:
        try:
            err = resp.json()["error"]
            return err.get("code", f"http_{resp.status_code}"), err.get("message", "")
        except (ValueError, KeyError, TypeError, AttributeError):
            return f"http_{resp.status_code}", ""

    @staticmethod
    def _backoff(attempt: int) -> float:
        return 2 ** (attempt - 1) + random.uniform(0, 0.5)

    def _delay(self, resp: httpx.Response, attempt: int) -> float | None:
        """Seconds to wait before a retry, or None to stop retrying."""
        header = resp.headers.get("Retry-After")
        if header is None:
            return self._backoff(attempt)
        try:
            seconds = max(float(header), 0.0)
        except ValueError:
            return self._backoff(attempt)
        return seconds if seconds <= MAX_WAIT else None

Four details matter here. The reservation happens before the request, so a capped run never touches the network, and parallel calls can't all pass the check at once. The loop returns strings for every failure instead of raising, so the model reads "422 invalid_parameters: …" and can correct its next call. Retry-After wins over the backoff schedule when it's there, and a wait longer than MAX_WAIT ends the retries rather than stalling the agent. And project keeps only fields that appear in the recorded responses before it applies MAX_CHARS.

Why project at all? The recorded Search response carries redirect_link tokens on each result, and the recorded News response carries thumbnail, story_token and a menu_links list. Those cost context and tell the model nothing. Cutting raw JSON at a character count also lands mid-field. project drops whole results instead, so the output stays valid JSON. The trade-off is that fields outside KEEP, such as related_questions or local_results, never reach the model; add the ones your agent needs. The recorded sample is from the Search endpoint, and we haven't recorded Search Light, so if its response has neither organic_results nor news_results, project falls back to the raw text cut at MAX_CHARS. Check the shape in the playground and adjust RESULT_KEYS if it differs.

Step 4: How do you turn the client into LangChain tools?

Wrap the client in a factory so each run gets its own budget. Inside, decorate one function per endpoint with @tool, pass the args_schema, and write a docstring that tells the model when to pick this tool over the other. The docstring is the tool description, so it's the most important text in the file after the schema.

from langchain.tools import tool


def build_tools(budget: CreditBudget):
    client = PerceptlayerClient(budget)

    @tool("google_search", args_schema=SearchInput)
    def google_search(
        query: str,
        num: Optional[int] = None,
        page: Optional[int] = None,
        hl: Optional[str] = None,
        gl: Optional[str] = None,
        safe: Optional[str] = None,
        sort: Optional[str] = None,
    ) -> str:
        """Search Google and return the results as JSON text. Use for general
        questions, documentation and anything that is not breaking news."""
        return client.get(
            "/v1/google-search/search-light",
            {"query": query, "num": num, "page": page, "hl": hl,
             "gl": gl, "safe": safe, "sort": sort},
        )

    @tool("google_news", args_schema=NewsInput)
    def google_news(
        q: Optional[str] = None,
        gl: Optional[str] = None,
        hl: Optional[str] = None,
        page: Optional[int] = None,
    ) -> str:
        """Search Google News and return recent articles as JSON text. Use for
        current events and anything where recency matters."""
        return client.get(
            "/v1/google-news/news",
            {"q": q, "gl": gl, "hl": hl, "page": page},
        )

    return [google_search, google_news]

You can test a tool without any model. Call build_tools(CreditBudget(cap=5))[1].invoke({"q": "technology"}) and you get a JSON string: the projected results, not the raw response. Pass {"query": "x" * 3000} to the search tool and the schema rejects it before a request is sent.

What does the model actually see? Here is the recorded News response for q=technology, trimmed to two results. The bulky fields are marked with ….

{
  "news_results": [
    {
      "position": 1,
      "title": "Top Strategic Technology Trends for 2027",
      "link": "https://www.gartner.com/en/articles/top-technology-trends-2027",
      "snippet": "Explore Gartner's Top Strategic Technology Trends for 2027 — 10 trends that will shape autonomy, resilience and governance in the years...",
      "iso_date": "2026-10-20T15:59:45+00:00",
      "thumbnail": "https://www.gartner.com/ngw/globalassets/…/itsym-top-tech-trends.webp",
      "source": { "name": "Gartner", "icon": "…", "title": "Gartner" },
      "story_token": "CBMib0FVX3lxTE1zMGNN…",
      …
    },
    …
  ],
  "menu_links": [ { "title": "U.S.", "link": "…", "topic_token": "…" }, … ]
}

After project, the news tool returns this instead, with thumbnail, story_token, menu_links and the nested source object gone:

[{"position": 1, "title": "Top Strategic Technology Trends for 2027", "link": "https://www.gartner.com/en/articles/top-technology-trends-2027", "snippet": "Explore Gartner's Top Strategic Technology Trends for 2027 — 10 trends that…", "source": "Gartner", "iso_date": "2026-10-20T15:59:45+00:00"}, …]

The recorded Search response for query=best restaurants in Berlin works the same way. Each item in organic_results has position, title, link, snippet and source, plus a long redirect_link that the projection drops:

{
  "organic_results": [
    {
      "position": 1,
      "title": "The Best Restaurants in Berlin - Food Guide & Map",
      "link": "https://andershusa.com/best-restaurants-berlin-germany-guide/",
      "redirect_link": "https://www.google.com/goto?url=CAESegHrOzAV7vOTOiRY…",
      "snippet": "Dec 3, 2025 — Tim Raue is the most internationally renowned restaurant in Berlin. …",
      "source": "Anders & Kaitlin",
      …
    },
    …
  ],
  …
}

That recording is from /v1/google-search/search, a sibling of Search Light, so treat it as the shape to expect rather than a guarantee. To see a real Search Light or News response with your own parameters, run the call in the playground on the Search Light page or the News page.

Step 5: How do you run the agent with a credit cap?

Create one budget per run, build the tools from it, and hand them to the agent. create_agent is imported from langchain.agents, takes a "provider:model" string and a system_prompt, as the LangChain agents docs show. The model ID below, claude-sonnet-5-5, is the Claude Sonnet 5.5 API ID listed on Anthropic's models overview (2026); the agents docs use a different Claude example ID, so check the models page for the current one. Print the budget at the end so every run reports what it spent. The cap is enforced in the tools, not in the prompt, and each call reserves its credits under a lock before the request goes out, so neither a persuasive model nor parallel tool calls can push the run past it.

from langchain.agents import create_agent

MODEL = "anthropic:claude-sonnet-5-5"  # swap for any chat model LangChain supports


def run(question: str, max_credits: int = 10) -> str:
    budget = CreditBudget(cap=max_credits)
    agent = create_agent(
        MODEL,
        tools=build_tools(budget),
        system_prompt=(
            "Answer using the tools. Use google_news for recent events and "
            "google_search otherwise. If a tool says the credit cap is "
            "reached, stop calling tools and answer with what you have."
        ),
    )
    result = agent.invoke({"messages": [{"role": "user", "content": question}]})
    print(f"Credits used this run: {budget.spent} of {budget.cap}")
    return result["messages"][-1].content


if __name__ == "__main__":
    print(run("What changed in the news about AI agent frameworks this week?"))

What the run prints is two things: the model's final answer, which depends on the model and on whatever the news is that day, and one line in the form Credits used this run: N of 10, where N is the number of successful calls at 1 credit each, plus any call whose reply was lost. We haven't recorded an agent run, so we won't show an answer here. If the agent loops, the first call past the cap comes back with the cap message instead of data, and the run ends at your ceiling. The credit line prints on every run, which is also a cheap way to spot a prompt that makes the agent over-search.

When should you use MCP instead of a custom tool?

Use a custom tool when you want control: typed arguments you chose, your own retry rules and a budget in code. Use the MCP server when you want breadth. It exposes search_endpoints, get_endpoint and call_endpoint at https://api.perceptlayer.com/mcp, so an agent can discover endpoints across the whole catalogue of 248 websites and 1,659 endpoints without you writing a wrapper for each.

The two compose well. Searching the catalogue and reading endpoint details cost nothing; call_endpoint spends the endpoint's credits on success only. A custom tool is the better fit when the agent needs one or two sources and you want predictable arguments. MCP fits an agent that has to pick its own sources. The MCP docs and the Claude MCP search tutorial cover that route.

Your next step: get a free API key, paste the code into one file, and run it with a cap of 5 credits. Then browse the full list of APIs and add a second wrapper for the next source your agent needs.

langchainpythontool callinggoogle searchai agents