如何为 CrewAI 创建 Scavio 搜索工具

import os, requests
from crewai.tools import BaseTool
from typing import Optional

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

class ScavioSearchTool(BaseTool):
    name: str = 'scavio_search'
    description: str = 'Search the web for current information. Supports platforms: google, reddit, youtube, amazon, walmart. Default is google. Returns structured results with titles, links, and snippets.'

    def _run(self, query: str, platform: Optional[str] = None) -> str:
        body = {'query': query, 'country_code': 'us'}
        if platform and platform != 'google':
            body['platform'] = platform
        data = requests.post('https://api.scavio.dev/api/v1/search',
            headers=SH, json=body).json()
        results = data.get('organic_results', [])[:5]
        formatted = []
        for r in results:
            formatted.append(f"{r.get('position', 0)}. {r['title']}\n   {r.get('link', '')}\n   {r.get('snippet', '')[:120]}")
        return f'Results for "{query}" ({platform or "google"}):\n' + '\n\n'.join(formatted)

search_tool = ScavioSearchTool()
print(search_tool._run('best python framework 2026')[:200])

from crewai import Agent, Task, Crew

researcher = Agent(
    role='Senior Researcher',
    goal='Find comprehensive, current data on the given topic using web search across multiple platforms',
    backstory='You are an expert researcher who searches Google for official sources, Reddit for user opinions, and YouTube for tutorial coverage.',
    tools=[search_tool],
    verbose=True
)

analyst = Agent(
    role='Data Analyst',
    goal='Analyze research findings and produce actionable insights with data-backed recommendations',
    backstory='You synthesize data from multiple sources into clear, honest analysis. You note when data is limited or uncertain.',
    tools=[],
    verbose=True
)

print('Agents defined: Researcher (with search tool) + Analyst')

def run_research_crew(topic):
    research_task = Task(
        description=f'Research "{topic}" thoroughly. Search Google for top results, then search Reddit for real user opinions. Return raw findings organized by source.',
        expected_output='Structured research findings from Google and Reddit with links and key quotes.',
        agent=researcher
    )
    analysis_task = Task(
        description=f'Analyze the research findings on "{topic}". Identify consensus opinions, areas of disagreement, and produce a ranked recommendation with honest tradeoffs.',
        expected_output='Ranked analysis with pros/cons and a clear recommendation. Note any data gaps.',
        agent=analyst
    )
    crew = Crew(
        agents=[researcher, analyst],
        tasks=[research_task, analysis_task],
        verbose=True
    )
    result = crew.kickoff()
    return result

result = run_research_crew('best SERP API for Python developers')
print(f'\n--- Final Output ---\n{str(result)[:500]}')

def multi_platform_crew(topic):
    google_task = Task(
        description=f'Search Google for "{topic}" and list the top 5 results with titles, URLs, and key points from snippets.',
        expected_output='Top 5 Google results with summaries.',
        agent=researcher
    )
    reddit_task = Task(
        description=f'Search Reddit for "{topic}" to find real user discussions, complaints, and recommendations. Quote specific user opinions.',
        expected_output='Reddit discussion summary with quoted opinions.',
        agent=researcher
    )
    synthesis_task = Task(
        description='Synthesize Google authority sources and Reddit user opinions into a balanced recommendation. Flag where official claims differ from user experience.',
        expected_output='Balanced analysis contrasting official sources with real user experience.',
        agent=analyst
    )
    crew = Crew(agents=[researcher, analyst],
                tasks=[google_task, reddit_task, synthesis_task], verbose=True)
    result = crew.kickoff()
    print(f'Crew complete. Search cost estimate: ~$0.020-0.040')
    return result

multi_platform_crew('best SERP API 2026')