# Directory Structure
```
├── .gitignore
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.txt
├── smithery.yaml
├── src
│ ├── client.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
*.egg-info/
.venv
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Brave Search MCP Server
[](https://smithery.ai/server/@arben-adm/brave-mcp-search)
This project implements a Model Context Protocol (MCP) server for Brave Search, allowing integration with AI assistants like Claude.
## Prerequisites
- Python 3.11+
- [uv](https://github.com/astral-sh/uv) - A fast Python package installer and resolver
## Installation
### Installing via Smithery
To install Brave Search MCP server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@arben-adm/brave-mcp-search):
```bash
npx -y @smithery/cli install @arben-adm/brave-mcp-search --client claude
```
### Manual Installation
1. Clone the repository:
```
git clone https://github.com/your-username/brave-search-mcp.git
cd brave-search-mcp
```
2. Create a virtual environment and install dependencies using uv:
```
uv venv
source .venv/bin/activate # On Windows, use: .venv\Scripts\activate
uv pip install -r requirements.txt
```
3. Set up your Brave Search API key:
```
export BRAVE_API_KEY=your_api_key_here
```
On Windows, use: `set BRAVE_API_KEY=your_api_key_here`
## Usage
1. Configure your MCP settings file (e.g., `claude_desktop_config.json`) to include the Brave Search MCP server:
```json
{
"mcpServers": {
"brave-search": {
"command": "uv",
"args": [
"--directory",
"path-to\\mcp-python\\brave-mcp-search\\src",
"run",
"server.py"
],
"env": {
"BRAVE_API_KEY": "YOUR_BRAVE_API_KEY_HERE"
}
}
}
}
```
Replace `YOUR_BRAVE_API_KEY_HERE` with your actual Brave API key.
2. Start the Brave Search MCP server by running your MCP-compatible AI assistant with the updated configuration.
3. The server will now be running and ready to accept requests from MCP clients.
4. You can now use the Brave Search functionality in your MCP-compatible AI assistant (like Claude) by invoking the available tools.
## Available Tools
The server provides two main tools:
1. `brave_web_search`: Performs a web search using the Brave Search API.
2. `brave_local_search`: Searches for local businesses and places.
Refer to the tool docstrings in `src/server.py` for detailed usage information.
## Development
To make changes to the project:
1. Modify the code in the `src` directory as needed.
2. Update the `requirements.txt` file if you add or remove dependencies:
```
uv pip freeze > requirements.txt
```
3. Restart the server to apply changes.
## Troubleshooting
If you encounter any issues:
1. Ensure your Brave API key is correctly set.
2. Check that all dependencies are installed.
3. Verify that you're using a compatible Python version.
4. If you make changes to the code, make sure to restart the server.
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp
httpx
fastmcp
[dev]
pytest
black
isort
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- braveApiKey
properties:
braveApiKey:
type: string
description: The API key for the Brave Search server.
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({command:'uv',args:['run', 'src/server.py'],env:{BRAVE_API_KEY:config.braveApiKey}})
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "brave-search-mcp"
version = "0.1.0"
description = "A Model Context Protocol (MCP) server for Brave Search"
authors = [
{name = "Your Name", email = "[email protected]"},
]
dependencies = [
"mcp",
"httpx",
"fastmcp",
]
requires-python = ">=3.11"
readme = "README.md"
license = {text = "MIT"}
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
packages = ["src"]
[project.optional-dependencies]
dev = [
"pytest",
"black",
"isort",
]
[tool.black]
line-length = 88
target-version = ['py37']
[tool.isort]
profile = "black"
line_length = 88
[tool.pytest.ini_options]
testpaths = ["tests"]
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
# Use a Python image with uv pre-installed
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
# Install the project into /app
WORKDIR /app
# Enable bytecode compilation
ENV UV_COMPILE_BYTECODE=1
# Copy from the cache instead of linking since it's a mounted volume
ENV UV_LINK_MODE=copy
# Install the project's dependencies using the lockfile and settings
RUN --mount=type=cache,target=/root/.cache/uv --mount=type=bind,source=uv.lock,target=uv.lock --mount=type=bind,source=pyproject.toml,target=pyproject.toml uv sync --frozen --no-install-project --no-dev --no-editable
# Then, add the rest of the project source code and install it
# Installing separately from its dependencies allows optimal layer caching
ADD . /app
RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev --no-editable
FROM python:3.12-slim-bookworm
WORKDIR /app
COPY --from=uv /root/.local /root/.local
COPY --from=uv --chown=app:app /app/.venv /app/.venv
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
# when running the container, add --db-path and a bind mount to the host's db file
ENTRYPOINT ["uv", "run", "src/server.py"]
```
--------------------------------------------------------------------------------
/src/client.py:
--------------------------------------------------------------------------------
```python
import asyncio
import logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from rich.console import Console
from rich.logging import RichHandler
from typing import Optional, Dict, Any
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
datefmt="[%X]",
handlers=[RichHandler()]
)
class BraveSearchClient:
def __init__(
self,
server_path: str,
api_key: str,
console: Optional[Console] = None
):
self.server_params = StdioServerParameters(
command="python",
args=[server_path],
env={"BRAVE_API_KEY": api_key}
)
self.console = console or Console()
self.logger = logging.getLogger("brave-search-client")
def _is_complex_query(self, query: str) -> bool:
"""Determine if a query is complex based on its characteristics"""
indicators = [
" and ", " or ", " why ", " how ", " what ", " explain ",
"compare", "difference", "analysis", "describe"
]
return any(indicator in query.lower() for indicator in indicators) or len(query.split()) > 5
async def _execute_search(
self,
session: ClientSession,
tool: str,
params: Dict[str, Any]
) -> str:
try:
# Adjust count based on query complexity
if "query" in params:
is_complex = self._is_complex_query(params["query"])
params["count"] = 20 if is_complex else 10
result = await session.call_tool(tool, params)
if result.is_error:
raise Exception(result.content[0].text)
return result.content[0].text
except Exception as e:
self.logger.error(f"Search failed: {str(e)}")
return f"Error: {str(e)}"
async def run_interactive(self):
"""Run interactive search client"""
try:
async with stdio_client(self.server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
self.console.print(
"Available tools:",
", ".join(tool.name for tool in tools)
)
while True:
query = self.console.input("\nSearch query (or 'quit'): ")
if query.lower() == "quit":
break
# Default to web search as it's used for answer formulation
search_type = "web"
is_complex = self._is_complex_query(query)
count = 20 if is_complex else 10
tool = "brave_web_search"
with self.console.status(f"Searching with {'complex' if is_complex else 'standard'} query..."):
result = await self._execute_search(
session,
tool,
{"query": query, "count": count}
)
self.console.print("\nResults:", style="bold green")
self.console.print(result)
except Exception as e:
self.logger.error(f"Client error: {str(e)}")
raise
if __name__ == "__main__":
import os
import sys
if len(sys.argv) < 2:
print("Usage: python client.py <path_to_server.py>")
sys.exit(1)
api_key = os.getenv("BRAVE_API_KEY")
if not api_key:
print("Error: BRAVE_API_KEY environment variable required")
sys.exit(1)
client = BraveSearchClient(sys.argv[1], api_key)
asyncio.run(client.run_interactive())
```
--------------------------------------------------------------------------------
/src/server.py:
--------------------------------------------------------------------------------
```python
from mcp.server.fastmcp import FastMCP
import httpx
import time
import asyncio
from typing import Optional, Dict, List, Any, Tuple
from dataclasses import dataclass
from enum import Enum
import os
import sys
import io
api_key = os.getenv("BRAVE_API_KEY")
if not api_key:
raise ValueError("BRAVE_API_KEY environment variable required")
class RateLimitError(Exception):
pass
@dataclass
class RateLimit:
per_second: int = 1
per_month: int = 2000
_requests: Dict[str, int] = None
_last_reset: float = 0.0
def __post_init__(self):
self._requests = {"second": 0, "month": 0}
self._last_reset = time.time()
def check(self):
now = time.time()
if now - self._last_reset > 1:
self._requests["second"] = 0
self._last_reset = now
if (self._requests["second"] >= self.per_second or
self._requests["month"] >= self.per_month):
raise RateLimitError("Rate limit exceeded")
self._requests["second"] += 1
self._requests["month"] += 1
class BraveSearchServer:
def __init__(self, api_key: str):
# Configure stdout for UTF-8
if sys.platform == 'win32':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
self.mcp = FastMCP(
"brave-search",
dependencies=["httpx", "asyncio"]
)
self.api_key = api_key
self.base_url = "https://api.search.brave.com/res/v1"
self.rate_limit = RateLimit()
self._client = None
self._setup_tools()
def get_client(self):
if self._client is None or self._client.is_closed:
self._client = httpx.AsyncClient(
headers={
"X-Subscription-Token": self.api_key,
"Accept": "application/json",
"Accept-Encoding": "gzip"
},
timeout=30.0
)
return self._client
async def _get_web_results(self, query: str, min_results: int) -> List[Dict]:
"""Fetch web results with pagination until minimum count is reached"""
client = self.get_client()
self.rate_limit.check()
try:
# Make a single request with the maximum allowed count
response = await client.get(
f"{self.base_url}/web/search",
params={
"q": query,
"count": min_results
}
)
response.raise_for_status()
data = response.json()
results = data.get("web", {}).get("results", [])
return results
except httpx.HTTPStatusError as e:
if e.response.status_code == 422:
# If we get a 422, try with a smaller count
response = await client.get(
f"{self.base_url}/web/search",
params={
"q": query,
"count": 10 # Fall back to smaller count
}
)
response.raise_for_status()
data = response.json()
return data.get("web", {}).get("results", [])
raise # Re-raise other HTTP errors
def _format_web_results(self, data: Dict, min_results: int = 10) -> str:
"""Format web search results with enhanced information"""
results = []
web_results = data.get("web", {}).get("results", [])
for result in web_results[:max(min_results, len(web_results))]:
# Strip or replace any potential Unicode characters
title = result.get('title', 'N/A').encode('ascii', 'replace').decode()
desc = result.get('description', 'N/A').encode('ascii', 'replace').decode()
formatted_result = [
f"Title: {title}",
f"Description: {desc}",
f"URL: {result.get('url', 'N/A')}"
]
# Add additional metadata if available
if "meta_url" in result:
formatted_result.append(f"Source: {result['meta_url']}")
if "age" in result:
formatted_result.append(f"Age: {result['age']}")
if "language" in result:
formatted_result.append(f"Language: {result['language']}")
results.append("\n".join(formatted_result))
return "\n\n".join(results)
def _setup_tools(self):
@self.mcp.tool()
async def brave_web_search(
query: str,
count: Optional[int] = 20
) -> str:
"""Execute web search using Brave Search API with improved results
Args:
query: Search terms
count: Desired number of results (10-20)
"""
min_results = max(10, min(count, 20)) # Ensure between 10 and 20
all_results = await self._get_web_results(query, min_results)
if not all_results:
return "No results found for the query."
formatted_results = []
for result in all_results[:min_results]:
formatted_result = [
f"Title: {result.get('title', 'N/A')}",
f"Description: {result.get('description', 'N/A')}",
f"URL: {result.get('url', 'N/A')}"
]
# Include additional context if available
if result.get('extra_snippets'):
formatted_result.append("Additional Context:")
formatted_result.extend([f"- {snippet}" for snippet in result['extra_snippets'][:2]])
formatted_results.append("\n".join(formatted_result))
return "\n\n".join(formatted_results)
@self.mcp.tool()
async def brave_local_search(
query: str,
count: Optional[int] = 20 # Changed default from 5 to 20
) -> str:
"""Search for local businesses and places
Args:
query: Location terms
count: Results (1-20
"""
self.rate_limit.check()
# Initial location search
params = {
"q": query,
"search_lang": "en",
"result_filter": "locations",
"count": 20 # Always request maximum results
}
client = self.get_client()
response = await client.get(
f"{self.base_url}/web/search",
params=params
)
response.raise_for_status()
data = response.json()
location_ids = self._extract_location_ids(data)
if not location_ids:
# If no local results found, fallback to web search
# with minimum 10 results
return await brave_web_search(query, 20)
# If we have less than 10 location IDs, try to get more
offset = 0
while len(location_ids) < 10 and offset < 40:
offset += 20
additional_response = await client.get(
f"{self.base_url}/web/search",
params={
"q": query,
"search_lang": "en",
"result_filter": "locations",
"count": 20,
"offset": offset
}
)
additional_data = additional_response.json()
location_ids.extend(self._extract_location_ids(additional_data))
# Get details for at least 10 locations
pois, descriptions = await self._get_location_details(
location_ids[:max(10, len(location_ids))]
)
return self._format_local_results(pois, descriptions)
async def _get_location_details(
self,
ids: List[str]
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Fetch POI and description data for locations"""
client = self.get_client()
pois_response, desc_response = await asyncio.gather(
client.get(
f"{self.base_url}/local/pois",
params={"ids": ids}
),
client.get(
f"{self.base_url}/local/descriptions",
params={"ids": ids}
)
)
return (
pois_response.json(),
desc_response.json()
)
def _extract_location_ids(self, data: Dict) -> List[str]:
"""Extract location IDs from search response"""
return [
result["id"]
for result in data.get("locations", {}).get("results", [])
if "id" in result
]
def _format_local_results(
self,
pois: Dict[str, Any],
descriptions: Dict[str, Any]
) -> str:
"""Format local search results with details"""
results = []
for poi in pois.get("results", []):
location = {
"name": poi.get("name", "N/A"),
"address": self._format_address(poi.get("address", {})),
"phone": poi.get("phone", "N/A"),
"rating": self._format_rating(poi.get("rating", {})),
"price": poi.get("priceRange", "N/A"),
"hours": ", ".join(poi.get("openingHours", [])) or "N/A",
"description": descriptions.get("descriptions", {}).get(
poi["id"], "No description available"
)
}
results.append(
f"Name: {location['name']}\n"
f"Address: {location['address']}\n"
f"Phone: {location['phone']}\n"
f"Rating: {location['rating']}\n"
f"Price Range: {location['price']}\n"
f"Hours: {location['hours']}\n"
f"Description: {location['description']}"
)
return "\n---\n".join(results) or "No local results found"
def _format_address(self, addr: Dict) -> str:
"""Format address components"""
components = [
addr.get("streetAddress", ""),
addr.get("addressLocality", ""),
addr.get("addressRegion", ""),
addr.get("postalCode", "")
]
return ", ".join(filter(None, components)) or "N/A"
def _format_rating(self, rating: Dict) -> str:
"""Format rating information"""
if not rating:
return "N/A"
# Use ASCII star (*) instead of Unicode star
stars = "*" * int(float(rating.get('ratingValue', 0)))
return f"{rating.get('ratingValue', 'N/A')} {stars} ({rating.get('ratingCount', 0)} reviews)"
def run(self):
"""Start the MCP server"""
self.mcp.run()
if __name__ == "__main__":
server = BraveSearchServer(api_key)
server.run()
```