Shane
045ddabac7
- agent/web_mcp.py: stdio MCP server exposing web_search and web_scrape, backed by the self-hosted Firecrawl stack on xNAS (no API key needed) - agent.py: Agent now attaches mcp_servers built from config; EXTRA_MCP_SERVERS env var allows adding arbitrary HTTP/SSE MCP servers as JSON - Dockerfile: installs livekit-agents[mcp], copies web_mcp.py - .env.example: WEB_MCP_ENABLED, FIRECRAWL_BASE, EXTRA_MCP_SERVERS documented
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""Web access MCP server — exposes Firecrawl search + scrape as MCP tools.
|
|
|
|
Runs over stdio inside the voice container. The agent attaches it via
|
|
MCPServerStdio, so the LLM can call web_search / web_scrape during a
|
|
conversation to look things up in real time.
|
|
|
|
Backed by the self-hosted Firecrawl stack on xNAS (no API key needed):
|
|
- POST {FIRECRAWL_BASE}/v1/search -> ranked results with title/description
|
|
- POST {FIRECRAWL_BASE}/v1/scrape -> page content as markdown
|
|
|
|
Config (env vars, all optional):
|
|
FIRECRAWL_BASE default http://192.168.86.2:3002
|
|
FIRECRAWL_API_KEY only needed if the Firecrawl instance requires auth
|
|
WEB_SEARCH_LIMIT default 5 results per search
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
FIRECRAWL_BASE = os.environ.get("FIRECRAWL_BASE", "http://192.168.86.2:3002").rstrip("/")
|
|
FIRECRAWL_API_KEY = os.environ.get("FIRECRAWL_API_KEY", "")
|
|
WEB_SEARCH_LIMIT = int(os.environ.get("WEB_SEARCH_LIMIT", "5"))
|
|
|
|
mcp = FastMCP("web-access")
|
|
|
|
|
|
def _headers() -> dict[str, str]:
|
|
h = {"Content-Type": "application/json"}
|
|
if FIRECRAWL_API_KEY:
|
|
h["Authorization"] = f"Bearer {FIRECRAWL_API_KEY}"
|
|
return h
|
|
|
|
|
|
@mcp.tool()
|
|
async def web_search(query: str, limit: int | None = None) -> str:
|
|
"""Search the web and return ranked results with titles, URLs, and descriptions.
|
|
|
|
Use this to find current information, news, facts, or sources about a topic.
|
|
Returns a compact text summary — not raw JSON.
|
|
"""
|
|
limit = min(limit or WEB_SEARCH_LIMIT, 10)
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.post(
|
|
f"{FIRECRAWL_BASE}/v1/search",
|
|
headers=_headers(),
|
|
json={"query": query, "limit": limit},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
if not data.get("success"):
|
|
return f"Search failed: {data.get('error', 'unknown error')}"
|
|
|
|
results = data.get("data", [])
|
|
if not results:
|
|
return f"No results found for: {query}"
|
|
|
|
lines = [f"Web search results for: {query}", ""]
|
|
for i, r in enumerate(results, 1):
|
|
title = (r.get("title") or "(no title)").strip()
|
|
url = (r.get("url") or "").strip()
|
|
desc = (r.get("description") or "").strip().replace("\n", " ")
|
|
lines.append(f"{i}. {title}")
|
|
if desc:
|
|
lines.append(f" {desc[:300]}")
|
|
if url:
|
|
lines.append(f" URL: {url}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
async def web_scrape(url: str, max_chars: int = 8000) -> str:
|
|
"""Fetch a web page and return its content as readable markdown.
|
|
|
|
Use this after web_search to read the full content of a promising result.
|
|
The content is truncated to max_chars (default 8000) to stay within context.
|
|
"""
|
|
async with httpx.AsyncClient(timeout=90) as client:
|
|
resp = await client.post(
|
|
f"{FIRECRAWL_BASE}/v1/scrape",
|
|
headers=_headers(),
|
|
json={"url": url, "formats": ["markdown"], "timeout": 60000},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
if not data.get("success"):
|
|
return f"Scrape failed: {data.get('error', 'unknown error')}"
|
|
|
|
result = data.get("data", {})
|
|
markdown = (result.get("markdown") or "").strip()
|
|
if not markdown:
|
|
return f"No content extracted from {url}"
|
|
|
|
title = (result.get("metadata", {}).get("title") or "").strip()
|
|
header = f"Page: {title}\nURL: {url}\n\n" if title else f"URL: {url}\n\n"
|
|
if len(markdown) > max_chars:
|
|
markdown = markdown[:max_chars] + "\n\n[content truncated]"
|
|
return header + markdown
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="stdio")
|