Project Goal
In 30 minutes, build a LangChain Agent that:
- Receives user questions
- Auto-determines if SERP search is needed
- Calls SERP API for real-time Google data
- Feeds results to LLM for answer generation
- Includes citation references
Step 1: Environment (2 minutes)
pip install langchain langchain-anthropic requests
Environment variables:
export SERPBASE_KEY="your-key"
export ANTHROPIC_KEY="your-key"
Step 2: Tool Definition (5 minutes)
import os
import requests
from langchain.tools import Tool
SERPBASE_KEY = os.environ["SERPBASE_KEY"]
ENDPOINT = "https://api.serpbase.dev/google/search"
def serp_search(query: str) -> str:
"""Search Google for real-time results. Returns top 5 organic + PAA + knowledge graph.
Args:
query: search query string
"""
r = requests.post(
ENDPOINT,
headers={"X-API-Key": SERPBASE_KEY},
json={"q": query, "gl": "us", "hl": "en", "num": 5},
timeout=10,
)
r.raise_for_status()
data = r.json()
parts = []
# Organic results
for i, item in enumerate(data.get("organic", []), 1):
parts.append(f"[{i}] {item['title']}\n{item['link']}\n{item.get('snippet', '')}")
# People Also Ask
paa = data.get("people_also_ask", [])
if paa:
parts.append("\nRelated Questions:")
for q in paa[:3]:
parts.append(f"- {q.get('question', q)}")
# Knowledge Graph
kg = data.get("knowledge_graph", {})
if kg:
parts.append(f"\nKnowledge Graph: {kg.get('title', '')} - {kg.get('description', '')}")
return "\n\n".join(parts)
serp_tool = Tool(
name="Google Search",
func=serp_search,
description="Search Google for real-time results, returns SERP data (title, URL, snippet, PAA, knowledge graph). Input is query string.",
)
Step 3: Agent Integration (8 minutes)
from langchain_anthropic import ChatAnthropic
from langchain.agents import initialize_agent, AgentType
llm = ChatAnthropic(
model="claude-sonnet-4-5",
anthropic_api_key=os.environ["ANTHROPIC_KEY"],
)
agent = initialize_agent(
tools=[serp_tool],
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
max_iterations=3,
handle_parsing_errors=True,
)
def ask(question: str) -> str:
return agent.run(question)
Step 4: Test (5 minutes)
# Test 1: Simple query
print(ask("2026 SERP API latest pricing?"))
# Test 2: Multi-step query
print(ask("Compare SerpBase and SerpApi price and speed"))
# Test 3: Knowledge graph query
print(ask("What is Claude?"))
Step 5: Add Streaming (5 minutes)
import asyncio
from langchain.agents import AgentExecutor
async def ask_streaming(question: str):
result = agent.astream(question)
async for event in result:
if "values" in event:
print(event["values"], end="", flush=True)
if "messages" in event:
for msg in event["messages"]:
print(f"\n[Tool] {msg.content}", end="", flush=True)
# Usage
asyncio.run(ask_streaming("2026 SERP API cheapest?"))
Step 6: Add Citation Parsing (5 minutes)
import re
def parse_citations(answer: str) -> list:
"""Extract [1] [2] references from answer"""
pattern = r"\[(\d+)\]"
return [int(m) for m in re.findall(pattern, answer)]
def render_with_citations(answer: str, sources: list) -> str:
"""Replace [1] with clickable links"""
citations = parse_citations(answer)
if not citations:
return answer
source_list = "\n".join(
f"[{i+1}] {s['title']} - {s['link']}"
for i, s in enumerate(sources)
)
return f"{answer}\n\n**Sources:**\n{source_list}"
Complete Demo
def full_demo():
question = "2026 SERP API price and speed comparison"
answer = ask(question)
print(answer)
# Show sources (optional)
sources = [
{"title": "SERP API Comparison 2026", "link": "https://example.com/1"},
{"title": "SERP API Benchmark", "link": "https://example.com/2"},
]
print("\n" + render_with_citations(answer, sources))
if __name__ == "__main__":
full_demo()
4 Key Design Points
1. Tool Description Should Be Detailed
# Wrong: description="Search"
# Right: description="Search Google for real-time results, returns SERP data (title, URL, snippet, PAA, knowledge graph). Input is query string."
2. max_iterations Not Too High
# Wrong: max_iterations=10 (may infinite loop)
# Right: max_iterations=3 (3 is enough)
3. handle_parsing_errors
# Must set True, otherwise LLM output format errors can freeze
agent = initialize_agent(..., handle_parsing_errors=True)
4. Cache LLM Results
import hashlib
import json
CACHE = {}
def ask_cached(question):
key = hashlib.md5(question.encode()).hexdigest()
if key in CACHE:
return CACHE[key]
answer = ask(question)
CACHE[key] = answer
return answer
5 Real Workflows
1. Customer Support Bot
def customer_support(question):
sources = serp_search(f"{question} site:mysite.com")
return ask(f"Based on these sources, answer: {question}\n\nSources:{sources}")
2. Research Assistant
def research(question):
return ask(question)
# Agent auto-decides if SERP is needed
3. SEO Content Generation
def seo_article(keyword):
sources = serp_search(keyword)
prompt = f"Based on these top 5 Google SERP results, write a 1000-word SEO article:\n{sources}"
return ask(prompt)
4. Competitor Monitoring
def competitor_monitor(brand, competitors):
for comp in competitors:
sources = serp_search(f"{brand} vs {comp}")
return ask(f"Analyze competitor {comp} vs {brand}:\n{sources}")
5. AI Assistant
def ai_assistant(question):
return ask(question)
# Agent auto-decides if SERP is needed
Cost Analysis (Per Call)
| Item | Cost |
|---|---|
| SerpBase | $0.0003 |
| Claude Sonnet input (2k tokens) | $0.006 |
| Claude Sonnet output (500 tokens) | $0.0075 |
| Total | $0.0138 |
100 calls/day = $1.38/day = $41/month. Switch to Haiku to reduce to $0.005/call.
5 Common Mistakes
- Tool description too short: LLM doesn't know when to call
- max_iterations=0 or too high: either no calls or infinite loop
- No handle_parsing_errors: LLM output format errors freeze
- No max_iterations: may infinite loop
- No caching for repeated questions: high cost, slow
Deployment Options
| Method | Best for |
|---|---|
| Local script | Personal / testing |
| FastAPI + Docker | Production web app |
| Cloud Run / Lambda | Serverless, auto-scale |
| Streamlit / Gradio | Demo showcase |
100 free searches: serpbase.dev signup, small-scale Agent test.
Top comments (0)