Web Search & Grounding Skill

You are an expert in building Perplexity-style search-augmented generation pipelines,

combining live web retrieval with LLM synthesis for grounded, cited research responses.

You write production-ready search pipelines in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — Tavily: Best LLM-Optimised Search

# pip install tavily-python

from tavily import TavilyClient
import os

tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))


def tavily_search(
    query: str,
    search_depth: str = "advanced",   # "basic" (fast) or "advanced" (thorough)
    max_results: int = 5,
    include_domains: list[str] | None = None,
    exclude_domains: list[str] | None = None,
    include_raw_content: bool = False
) -> dict:
    """
    Tavily search — purpose-built for LLM agents.
    Returns clean, structured results (not raw HTML).
    """
    results = tavily.search(
        query=query,
        search_depth=search_depth,
        max_results=max_results,
        include_domains=include_domains or [],
        exclude_domains=exclude_domains or [],
        include_raw_content=include_raw_content,
        include_answer=True          # AI-generated summary of results
    )
    return results


def tavily_imi_search(query: str) -> dict:
    """
    IMI-tuned search: focus on sports, fan behaviour, sponsorship news.
    Excludes low-quality sources.
    """
    return tavily_search(
        query=query,
        search_depth="advanced",
        max_results=7,
        include_domains=[
            "bbc.co.uk", "theguardian.com", "sportspromedia.com",
            "twobirds.com", "nielsen.com", "statista.com",
            "marketresearch.com", "deloitte.com"
        ],
        exclude_domains=["reddit.com", "quora.com"]
    )


def format_search_results(results: dict) -> str:
    """Format Tavily results for LLM context injection."""
    sources = results.get("results", [])
    formatted = []
    for i, source in enumerate(sources, 1):
        formatted.append(
            f"[{i}] {source.get('title', 'No title')}\n"
            f"URL: {source.get('url', '')}\n"
            f"Content: {source.get('content', '')[:500]}"
        )
    return "\n\n".join(formatted)

Part 2 — Perplexity-Style Grounded Generation

import anthropic, re

def perplexity_style_answer(
    query: str,
    search_fn=None,
    model: str = "claude-3-5-sonnet-20241022"
) -> dict:
    """
    Full Perplexity-style pipeline:
    1. Search web for live results
    2. Inject as context
    3. Generate answer with inline citations [1][2][3]
    4. Return answer + sources
    """
    client = anthropic.Anthropic()
    if search_fn is None:
        search_fn = tavily_imi_search

    # 1. Search
    results = search_fn(query)
    context = format_search_results(results)
    sources = [{"index": i+1, "url": r["url"], "title": r.get("title", "")}
               for i, r in enumerate(results.get("results", []))]

    # 2. Generate with citations
    response = client.messages.create(
        model=model,
        max_tokens=1000,
        system=(
            "You are an IMI sports fan intelligence research assistant with access to live web search. "
            "Answer questions using the provided search results. "
            "Always cite sources using [1], [2], etc. inline. "
            "Use British English. Be specific and data-driven."
        ),
        messages=[{
            "role": "user",
            "content": (
                f"Search results:\n\n{context}\n\n"
                f"Question: {query}\n\n"
                "Provide a comprehensive answer with inline citations."
            )
        }]
    )

    answer = response.content[0].text

    return {
        "answer": answer,
        "sources": sources,
        "query": query
    }


def extract_citations(answer: str) -> list[int]:
    """Extract citation indices from answer text."""
    return [int(n) for n in re.findall(r'\[(\d+)\]', answer)]

Part 3 — Brave Search API

# pip install requests

import requests, os


def brave_search(
    query: str,
    count: int = 10,
    freshness: str | None = None,   # "pd" (24h), "pw" (week), "pm" (month), "py" (year)
    country: str = "GB"
) -> list[dict]:
    """
    Brave Search API — privacy-focused, no Google tracking.
    Good for UK sports/sponsorship news with country=GB.
    """
    headers = {
        "Accept": "application/json",
        "Accept-Encoding": "gzip",
        "X-Subscription-Token": os.getenv("BRAVE_SEARCH_API_KEY")
    }
    params = {
        "q": query,
        "count": count,
        "country": country,
        "search_lang": "en",
        "ui_lang": "en-GB"
    }
    if freshness:
        params["freshness"] = freshness

    resp = requests.get(
        "https://api.search.brave.com/res/v1/web/search",
        headers=headers,
        params=params
    )
    resp.raise_for_status()
    data = resp.json()

    results = []
    for item in data.get("web", {}).get("results", []):
        results.append({
            "title": item.get("title"),
            "url": item.get("url"),
            "description": item.get("description", ""),
            "age": item.get("age", "")
        })
    return results


def brave_news_search(query: str, days: int = 7) -> list[dict]:
    """Search for recent news articles."""
    freshness_map = {1: "pd", 7: "pw", 30: "pm", 365: "py"}
    freshness = freshness_map.get(days, "pw")
    return brave_search(f"{query} news", freshness=freshness, country="GB")

Part 4 — Exa Semantic Search

# pip install exa-py

from exa_py import Exa
import os

exa = Exa(api_key=os.getenv("EXA_API_KEY"))


def exa_semantic_search(
    query: str,
    num_results: int = 5,
    use_autoprompt: bool = True,
    include_domains: list[str] | None = None,
    start_published_date: str | None = None   # ISO format: "2024-01-01"
) -> list[dict]:
    """
    Exa semantic search — finds pages by meaning, not just keywords.
    Better than SerpAPI for research-quality sources.
    """
    result = exa.search_and_contents(
        query=query,
        num_results=num_results,
        use_autoprompt=use_autoprompt,
        include_domains=include_domains,
        start_published_date=start_published_date,
        text=True,
        highlights=True
    )

    return [
        {
            "title": r.title,
            "url": r.url,
            "text": r.text[:500] if r.text else "",
            "highlights": r.highlights or [],
            "published_date": r.published_date
        }
        for r in result.results
    ]


def exa_find_similar(url: str, num_results: int = 5) -> list[dict]:
    """Find pages similar to a given URL — great for expanding research."""
    result = exa.find_similar_and_contents(url, num_results=num_results, text=True)
    return [{"title": r.title, "url": r.url, "text": (r.text or "")[:300]}
            for r in result.results]

Part 5 — Multi-Source Search Aggregator

from concurrent.futures import ThreadPoolExecutor, as_completed


def multi_source_search(
    query: str,
    sources: list[str] = ["tavily", "brave"],
    max_per_source: int = 5
) -> list[dict]:
    """
    Search multiple providers in parallel, deduplicate by URL.
    """
    def search_source(source: str) -> list[dict]:
        if source == "tavily":
            r = tavily_search(query, max_results=max_per_source)
            return [{"url": x["url"], "title": x.get("title",""), "content": x.get("content",""), "source": "tavily"}
                    for x in r.get("results", [])]
        elif source == "brave":
            items = brave_search(query, count=max_per_source)
            return [{"url": x["url"], "title": x["title"], "content": x["description"], "source": "brave"}
                    for x in items]
        elif source == "exa":
            items = exa_semantic_search(query, num_results=max_per_source)
            return [{"url": x["url"], "title": x["title"], "content": x["text"], "source": "exa"}
                    for x in items]
        return []

    all_results = []
    seen_urls = set()

    with ThreadPoolExecutor(max_workers=len(sources)) as executor:
        futures = {executor.submit(search_source, s): s for s in sources}
        for future in as_completed(futures):
            for item in future.result():
                if item["url"] not in seen_urls:
                    seen_urls.add(item["url"])
                    all_results.append(item)

    return all_results


def grounded_imi_research(topic: str) -> dict:
    """
    Full grounded research pipeline for IMI topics.
    Searches web → aggregates → generates cited answer.
    """
    results = multi_source_search(
        f"{topic} sports fans sponsorship UK",
        sources=["tavily", "brave"]
    )
    context = "\n\n".join([
        f"[{i+1}] {r['title']}\n{r['content'][:400]}"
        for i, r in enumerate(results[:8])
    ])

    import anthropic
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        system="You are an IMI sports fan intelligence researcher. Use British English. Cite sources [1][2] etc.",
        messages=[{"role": "user", "content": f"Sources:\n{context}\n\nResearch question: {topic}"}]
    )

    return {
        "answer": response.content[0].text,
        "sources": [{"index": i+1, "url": r["url"], "title": r["title"]} for i, r in enumerate(results[:8])]
    }

Output Standards

• Always cite: inline [1][2][3] citations for grounded outputs
• Freshness: always set freshness / start_published_date for news queries
• Source diversity: use at least 2 search providers for important research
• Country: GB for UK-focused sports/sponsorship research
• Deduplicate: always deduplicate by URL when aggregating multiple sources
• British English in all queries and generated responses

pip install

pip install tavily-python exa-py requests anthropic

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)