server.py•8.25 kB
#!/usr/bin/env python3
"""
Exa Websets MCP Server
A Model Context Protocol server for interacting with Exa's Websets API.
Allows creating and managing websets for AI-powered web search and data collection.
"""
import os
from typing import Optional, List, Dict, Any, Literal
from fastmcp import FastMCP
import httpx
from pydantic import BaseModel, Field
# Initialize FastMCP server
mcp = FastMCP("Exa Websets")
# Configuration
EXA_API_KEY = os.getenv("EXA_API_KEY", "ce182a39-be3e-49b1-bdb2-15986f534790")
EXA_BASE_URL = "https://api.exa.ai"
EXA_WEBSETS_URL = f"{EXA_BASE_URL}/websets/v0/websets"
class WebsetSearchConfig(BaseModel):
"""Configuration for webset search parameters"""
query: str = Field(..., min_length=1, max_length=5000, description="Natural language search query")
count: int = Field(default=10, ge=1, description="Number of items to find")
entity: Optional[str] = Field(default=None, description="Entity type (company, person, article, research_paper, custom)")
criteria: Optional[List[str]] = Field(default=None, description="List of criteria descriptions")
recall: bool = Field(default=False, description="Whether to provide recall estimates")
class WebsetCreateParams(BaseModel):
"""Parameters for creating a new webset"""
search: Optional[WebsetSearchConfig] = None
external_id: Optional[str] = Field(default=None, max_length=300, description="External identifier for the webset")
metadata: Optional[Dict[str, str]] = Field(default=None, description="Key-value metadata pairs")
async def make_exa_request(method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
"""Make a request to the Exa API"""
if endpoint == "/websets":
url = EXA_WEBSETS_URL
elif endpoint.startswith("/websets/"):
url = f"{EXA_WEBSETS_URL}{endpoint[8:]}" # Remove /websets prefix
else:
url = f"{EXA_BASE_URL}{endpoint}"
headers = {
"x-api-key": EXA_API_KEY,
"Content-Type": "application/json"
}
async with httpx.AsyncClient() as client:
if method.upper() == "POST":
response = await client.post(url, json=data, headers=headers)
elif method.upper() == "GET":
response = await client.get(url, headers=headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
return response.json()
@mcp.tool
async def search_exa(
query: str,
num_results: int = 10,
search_type: str = "neural",
include_domains: Optional[List[str]] = None,
exclude_domains: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Perform a basic Exa search (for testing API connectivity).
Args:
query: Search query
num_results: Number of results to return (default: 10)
search_type: Type of search (neural, keyword, or auto)
include_domains: List of domains to include
exclude_domains: List of domains to exclude
Returns:
Search results from Exa
"""
payload = {
"query": query,
"numResults": num_results,
"type": search_type
}
if include_domains:
payload["includeDomains"] = include_domains
if exclude_domains:
payload["excludeDomains"] = exclude_domains
try:
result = await make_exa_request("POST", "/search", payload)
return result
except httpx.HTTPStatusError as e:
return {"error": f"Search failed: {e.response.status_code} - {e.response.text}"}
@mcp.tool
async def create_webset(
query: str,
count: int = 10,
entity: Optional[str] = None,
external_id: Optional[str] = None,
criteria: Optional[List[str]] = None,
recall: bool = False,
metadata: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
Create a new Exa webset with search configuration.
Args:
query: Natural language search query describing what you're looking for
count: Number of items the webset will attempt to find (default: 10)
entity: Entity type (company, person, article, research_paper, custom)
external_id: External identifier for easier reference
criteria: List of criteria descriptions for evaluation
recall: Whether to provide recall estimates
metadata: Key-value metadata pairs
Returns:
Dictionary containing the created webset information
"""
# Build search configuration
search_config = {
"query": query,
"count": count
}
if entity:
if entity == "custom":
raise ValueError("Custom entity type requires a description")
search_config["entity"] = {"type": entity}
if criteria:
search_config["criteria"] = [{"description": desc} for desc in criteria]
if recall:
search_config["recall"] = True
# Build request payload
payload = {"search": search_config}
if external_id:
payload["externalId"] = external_id
if metadata:
payload["metadata"] = metadata
try:
result = await make_exa_request("POST", "/websets", payload)
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 409:
return {"error": "Webset with this externalId already exists"}
raise e
@mcp.tool
async def get_webset(webset_id: str) -> Dict[str, Any]:
"""
Get information about a specific webset.
Args:
webset_id: The unique identifier for the webset
Returns:
Dictionary containing webset information
"""
try:
result = await make_exa_request("GET", f"/{webset_id}")
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
return {"error": f"Webset {webset_id} not found"}
raise e
@mcp.tool
async def list_websets() -> Dict[str, Any]:
"""
List all websets in your account.
Returns:
Dictionary containing list of websets
"""
try:
result = await make_exa_request("GET", "")
return result
except httpx.HTTPStatusError as e:
return {"error": f"Failed to list websets: {e.response.status_code}"}
@mcp.tool
async def create_marketing_agencies_webset(
location: str = "US",
focus: str = "consumer products",
count: int = 10
) -> Dict[str, Any]:
"""
Create a webset to find marketing agencies based on location and focus area.
Args:
location: Geographic location (default: "US")
focus: Focus area or specialization (default: "consumer products")
count: Number of agencies to find (default: 10)
Returns:
Dictionary containing the created webset information
"""
query = f"Marketing agencies based in {location}, that focus on {focus}."
return await create_webset(
query=query,
count=count,
entity="company",
external_id=f"marketing-agencies-{location.lower()}-{focus.replace(' ', '-')}",
metadata={
"type": "marketing_agencies",
"location": location,
"focus": focus
}
)
@mcp.tool
async def create_tech_companies_webset(
location: str = "San Francisco",
stage: Optional[str] = None,
count: int = 10
) -> Dict[str, Any]:
"""
Create a webset to find tech companies based on location and optional stage.
Args:
location: Geographic location (default: "San Francisco")
stage: Company stage (e.g., "startup", "Series A", "public")
count: Number of companies to find (default: 10)
Returns:
Dictionary containing the created webset information
"""
query = f"Tech companies in {location}"
if stage:
query += f" that are {stage}"
return await create_webset(
query=query,
count=count,
entity="company",
external_id=f"tech-companies-{location.lower().replace(' ', '-')}",
metadata={
"type": "tech_companies",
"location": location,
"stage": stage or "any"
}
)
if __name__ == "__main__":
mcp.run()