Adding a search tool to a LangGraph research agent means creating a tool node in the state graph that calls a search API and routes results back into the agent's reasoning loop. LangGraph models agent workflows as state machines where nodes perform actions and edges define control flow. By adding a search node that calls the Scavio API and updating the agent state with structured results, you give the research agent the ability to gather live evidence, verify claims, and discover sources within its graph execution cycle.
Prerequisites
- Python 3.10+
- langgraph and langchain-core installed
- Scavio API key from scavio.dev
- Basic understanding of LangGraph state machines
Walkthrough
Step 1: Define the agent state
Create a TypedDict that holds the conversation messages, search results, and research status for the LangGraph state machine.
Python
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class ResearchState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: list[dict]
search_query: str
research_complete: boolStep 2: Create the search tool node
Build a node function that extracts the search query from state, calls the Scavio API, and updates the state with results.
Python
import os, requests
from langchain_core.messages import ToolMessage
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def search_node(state: ResearchState) -> dict:
query = state['search_query']
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': query, 'country_code': 'us'}).json()
results = [{
'title': r.get('title', ''),
'url': r.get('link', ''),
'snippet': r.get('snippet', ''),
} for r in data.get('organic_results', [])[:5]]
summary = '\n'.join([f"- {r['title']}: {r['snippet']}" for r in results])
return {
'search_results': results,
'messages': [ToolMessage(content=f'Search results for "{query}":\n{summary}',
tool_call_id='search')],
}Step 3: Build the reasoning node and router
Create the agent reasoning node that decides whether to search, continue researching, or finalize the answer.
Python
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, HumanMessage
llm = ChatOpenAI(model='gpt-4o')
def reasoning_node(state: ResearchState) -> dict:
system = ('You are a research agent. Analyze the conversation and decide:\n'
'1. If you need more data, respond with SEARCH: <query>\n'
'2. If you have enough data, respond with ANSWER: <your answer>')
messages = [HumanMessage(content=system)] + list(state['messages'])
response = llm.invoke(messages)
content = response.content
if content.startswith('SEARCH:'):
query = content.replace('SEARCH:', '').strip()
return {'messages': [response], 'search_query': query, 'research_complete': False}
else:
return {'messages': [response], 'research_complete': True}
def should_search(state: ResearchState) -> str:
if state.get('research_complete'):
return 'done'
return 'search'Step 4: Assemble and run the graph
Wire the nodes together into a LangGraph state machine with conditional edges and execute a research query.
Python
from langgraph.graph import StateGraph, END
def build_research_graph():
graph = StateGraph(ResearchState)
graph.add_node('reason', reasoning_node)
graph.add_node('search', search_node)
graph.add_conditional_edges('reason', should_search, {
'search': 'search',
'done': END,
})
graph.add_edge('search', 'reason')
graph.set_entry_point('reason')
return graph.compile()
research_agent = build_research_graph()
# Run a research task
result = research_agent.invoke({
'messages': [HumanMessage(content='What are the best vector databases for RAG in 2026?')],
'search_results': [],
'search_query': '',
'research_complete': False,
})
final_message = result['messages'][-1].content
print(final_message)Python Example
Python
import os, requests, operator
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
llm = ChatOpenAI(model='gpt-4o')
class ResearchState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
search_results: list[dict]
search_query: str
research_complete: bool
def search_node(state: ResearchState) -> dict:
q = state['search_query']
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': q, 'country_code': 'us'}).json()
results = [{'title': r.get('title',''), 'url': r.get('link',''),
'snippet': r.get('snippet','')} for r in data.get('organic_results',[])[:5]]
summary = '\n'.join([f"- {r['title']}: {r['snippet']}" for r in results])
return {'search_results': results,
'messages': [ToolMessage(content=f'Results for "{q}":\n{summary}', tool_call_id='search')]}
def reason_node(state: ResearchState) -> dict:
prompt = ('Research agent. Need more data? Say SEARCH: <query>. '
'Have enough? Say ANSWER: <response>')
msgs = [HumanMessage(content=prompt)] + list(state['messages'])
resp = llm.invoke(msgs)
if resp.content.startswith('SEARCH:'):
return {'messages': [resp], 'search_query': resp.content[7:].strip(),
'research_complete': False}
return {'messages': [resp], 'research_complete': True}
def router(state: ResearchState) -> str:
return 'done' if state.get('research_complete') else 'search'
graph = StateGraph(ResearchState)
graph.add_node('reason', reason_node)
graph.add_node('search', search_node)
graph.add_conditional_edges('reason', router, {'search': 'search', 'done': END})
graph.add_edge('search', 'reason')
graph.set_entry_point('reason')
agent = graph.compile()
result = agent.invoke({
'messages': [HumanMessage(content='Best vector databases for RAG in 2026?')],
'search_results': [], 'search_query': '', 'research_complete': False})
print(result['messages'][-1].content)JavaScript Example
JavaScript
// LangGraph.js research agent with Scavio search
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function searchNode(state) {
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H,
body: JSON.stringify({query: state.searchQuery, country_code: 'us'})
}).then(r => r.json());
const results = (data.organic_results || []).slice(0, 5).map(r => ({
title: r.title || '', url: r.link || '', snippet: r.snippet || ''
}));
const summary = results.map(r => \`- \${r.title}: \${r.snippet}\`).join('\n');
return {
searchResults: results,
messages: [...state.messages, {role: 'tool', content: \`Results: \${summary}\`}],
};
}
// Reasoning node calls your LLM to decide search vs answer
async function reasonNode(state) {
// Call LLM with state.messages, parse SEARCH:/ANSWER: prefix
// Return updated state with searchQuery or researchComplete
console.log(\`Processing \${state.messages.length} messages\`);
return state;
}
// Wire into LangGraph.js StateGraph
// const graph = new StateGraph({channels: {...}})
// graph.addNode('reason', reasonNode)
// graph.addNode('search', searchNode)
// graph.addConditionalEdges('reason', router, {search: 'search', done: END})
console.log('LangGraph research agent with search ready');Expected Output
JSON
Research agent execution:
reason -> SEARCH: vector databases RAG comparison 2026
search -> 5 results from Scavio
reason -> SEARCH: pinecone vs weaviate vs qdrant benchmarks
search -> 5 results from Scavio
reason -> ANSWER: The top vector databases for RAG in 2026 are...
Final answer includes cited sources from live search data.