Adding shared search capability to a multi-agent system lets each agent ground its reasoning in live web data without duplicating queries or blowing through API credits. The core pattern is a thin wrapper around a search API (like Scavio) combined with an in-memory or Redis-backed cache keyed by query hash. Each agent calls the shared tool, and the cache ensures identical queries across agents return instantly from the local store. This tutorial builds the shared search tool, wires it into CrewAI as an example, and shows how to adapt it for AutoGen or a custom orchestrator.
Prerequisites
- Python 3.10+ installed
- crewai and requests libraries installed (pip install crewai requests)
- A Scavio API key from scavio.dev
- Basic familiarity with multi-agent frameworks
Walkthrough
Step 1: Build the cached search tool
Create a search function with an in-memory cache keyed by the SHA-256 hash of the query and country. Any agent calling this function with the same query gets the cached result, saving credits and latency.
Python
import requests, os, hashlib, json
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
_cache = {}
def cached_search(query: str, country: str = 'us') -> dict:
key = hashlib.sha256(f'{query}:{country}'.encode()).hexdigest()
if key in _cache:
print(f'[cache hit] {query}')
return _cache[key]
resp = requests.post(ENDPOINT,
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
resp.raise_for_status()
data = resp.json()
_cache[key] = data
print(f'[cache miss] {query} -> {len(data.get("organic_results", []))} results')
return dataStep 2: Wrap as a CrewAI tool
CrewAI tools are decorated functions with a name and description. Wrap the cached search function so any CrewAI agent can call it.
Python
from crewai.tools import tool
@tool('web_search')
def web_search_tool(query: str) -> str:
'''Search the web for a query and return top 5 results with titles, links, and snippets.'''
data = cached_search(query)
results = data.get('organic_results', [])[:5]
lines = []
for r in results:
lines.append(f"Title: {r['title']}")
lines.append(f"URL: {r['link']}")
lines.append(f"Snippet: {r.get('snippet', '')}")
lines.append('')
return '\n'.join(lines)Step 3: Assign the tool to multiple agents
Create two agents that share the same search tool. When both agents research overlapping topics, the cache prevents duplicate API calls.
Python
from crewai import Agent, Task, Crew
researcher = Agent(
role='Market Researcher',
goal='Find the top 3 CRM tools by market share in 2026',
backstory='Expert in SaaS market analysis',
tools=[web_search_tool],
verbose=True
)
analyst = Agent(
role='Pricing Analyst',
goal='Compare pricing tiers of the top CRM tools',
backstory='Expert in SaaS pricing strategy',
tools=[web_search_tool],
verbose=True
)
task1 = Task(description='Research top CRM tools 2026', agent=researcher,
expected_output='List of top 3 CRM tools with market share')
task2 = Task(description='Compare CRM pricing tiers', agent=analyst,
expected_output='Pricing comparison table')
crew = Crew(agents=[researcher, analyst], tasks=[task1, task2], verbose=True)
result = crew.kickoff()
print(f'Cache size: {len(_cache)} queries (saved credits on duplicates)')Step 4: Adapt for AutoGen or custom agents
The same cached_search function works with any orchestrator. For AutoGen, register it as a function. For custom agents, call it directly. The cache is shared across all callers in the same process.
Python
# AutoGen example
from autogen import register_function
def autogen_search(query: str) -> str:
data = cached_search(query)
return json.dumps(data.get('organic_results', [])[:5], indent=2)
# Register for any AutoGen agent
# register_function(autogen_search, caller=assistant, executor=user_proxy,
# name='web_search', description='Search the web')
# Custom agent loop example
def agent_step(agent_name: str, query: str):
results = cached_search(query)
organic = results.get('organic_results', [])[:5]
print(f'[{agent_name}] {query} -> {len(organic)} results')
return organic
# Both agents share the cache
agent_step('researcher', 'best crm tools 2026')
agent_step('analyst', 'best crm tools 2026') # cache hitPython Example
Python
import requests, os, hashlib, json
from crewai import Agent, Task, Crew
from crewai.tools import tool
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
_cache = {}
def cached_search(query: str, country: str = 'us') -> dict:
key = hashlib.sha256(f'{query}:{country}'.encode()).hexdigest()
if key in _cache:
return _cache[key]
resp = requests.post(ENDPOINT,
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
resp.raise_for_status()
_cache[key] = resp.json()
return _cache[key]
@tool('web_search')
def web_search_tool(query: str) -> str:
'''Search the web and return top 5 results.'''
data = cached_search(query)
results = data.get('organic_results', [])[:5]
return '\n'.join(f"{r['title']}: {r['link']}" for r in results)
researcher = Agent(role='Researcher', goal='Find top CRM tools',
backstory='Market analyst', tools=[web_search_tool])
analyst = Agent(role='Analyst', goal='Compare CRM pricing',
backstory='Pricing expert', tools=[web_search_tool])
task1 = Task(description='Research CRM tools 2026', agent=researcher,
expected_output='Top 3 CRM tools')
task2 = Task(description='Compare pricing', agent=analyst,
expected_output='Pricing table')
crew = Crew(agents=[researcher, analyst], tasks=[task1, task2])
result = crew.kickoff()
print(f'Unique queries: {len(_cache)}')
print(result)JavaScript Example
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || 'your_scavio_api_key';
const ENDPOINT = 'https://api.scavio.dev/api/v1/search';
const cache = new Map();
async function cachedSearch(query, country = 'us') {
const key = `${query}:${country}`;
if (cache.has(key)) return cache.get(key);
const resp = await fetch(ENDPOINT, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: country })
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
cache.set(key, data);
return data;
}
async function agentSearch(agentName, query) {
const data = await cachedSearch(query);
const results = (data.organic_results || []).slice(0, 5);
console.log(`[${agentName}] ${query} -> ${results.length} results (cache size: ${cache.size})`);
return results;
}
async function main() {
await agentSearch('researcher', 'best crm tools 2026');
await agentSearch('analyst', 'best crm tools 2026');
await agentSearch('analyst', 'crm pricing comparison 2026');
console.log(`Total unique queries: ${cache.size}`);
}
main().catch(console.error);Expected Output
JSON
[cache miss] best crm tools 2026 -> 10 results
[cache hit] best crm tools 2026
[cache miss] crm pricing comparison 2026 -> 10 results
Total unique queries: 2
Researcher found: Salesforce, HubSpot, Pipedrive
Analyst pricing table:
Salesforce: $25-$300/user/mo
HubSpot: $0-$150/user/mo
Pipedrive: $14-$99/user/mo