Add live web search to an OpenAI Agents SDK agent

OpenAI Agents SDK agents can call Superhighway's five search tools via its built-in MCP support — or wrap the REST API as a plain @function_tool. Both paths work; the MCP path keeps the agent fully autonomous (no human billing loop).

Two paths

Path 1: MCP server (recommended for autonomous agents)

pip install openai-agents

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

async def main():
    async with MCPServerStdio(
        "npx",
        args=["-y", "superhighway-mcp"],
        env={
            "AGENT_PRIVATE_KEY": "0xYOUR_FUNDED_BASE_WALLET_KEY",
            "X402_NETWORK": "base",
        },
    ) as mcp:
        agent = Agent(
            name="researcher",
            instructions="You are a research assistant. Use web search to find current, accurate information.",
            mcp_servers=[mcp],
        )
        result = await Runner.run(agent, "What are the latest AI agent frameworks released this month?")
        print(result.final_output)

asyncio.run(main())

The agent gets all five tools: web_search, news_search, image_search, scrape, and research. Under the hood each call settles a $0.001–$0.005 USDC payment from the wallet — no API key, no human billing loop. See wallet setup or the x402 flow.

Path 2: REST API as a @function_tool (API key)

Get a free API key (1,000 calls/month) and wrap the REST endpoints as typed function tools:

pip install openai-agents httpx

import asyncio, os
import httpx
from agents import Agent, Runner, function_tool

API_KEY = os.environ["SUPERHIGHWAY_API_KEY"]  # from superhighway.walls.sh/pricing

@function_tool
def web_search(query: str, limit: int = 5) -> str:
    """Search the live web for current events, facts, and research."""
    r = httpx.get("https://superhighway.walls.sh/search",
        params={"q": query, "limit": limit},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10)
    r.raise_for_status()
    results = r.json()["results"]
    return "\n\n".join(f"**{x['title']}**\n{x['url']}\n{x['description']}" for x in results)

@function_tool
def news_search(query: str) -> str:
    """Search recent news articles."""
    r = httpx.get("https://superhighway.walls.sh/news",
        params={"q": query},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10)
    r.raise_for_status()
    results = r.json()["results"]
    return "\n\n".join(f"**{x['title']}** ({x.get('publishedDate','')})\n{x['url']}\n{x['description']}" for x in results)

@function_tool
def research(query: str) -> str:
    """Search + read the top pages for a query — returns full page content."""
    r = httpx.get("https://superhighway.walls.sh/research",
        params={"q": query, "pages": 2},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30)
    r.raise_for_status()
    data = r.json()
    return "\n\n---\n\n".join(f"## {p['title']}\n{p['url']}\n\n{p['markdown'][:3000]}"
                                  for p in data["pages"] if not p.get("error"))

async def main():
    agent = Agent(
        name="researcher",
        instructions="Use web_search, news_search, and research tools to answer questions with current information.",
        tools=[web_search, news_search, research],
    )
    result = await Runner.run(agent, "Summarize the top AI news from the last 48 hours.")
    print(result.final_output)

asyncio.run(main())

Which tools are available

Full API reference: /openapi.json. Get a free API key or learn the x402 wallet flow.