# Directory Structure
```
├── .gitignore
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.txt
├── smithery.yaml
├── src
│ ├── client.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | *.egg-info/
2 | .venv
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Brave Search MCP Server
2 |
3 | [](https://smithery.ai/server/@arben-adm/brave-mcp-search)
4 |
5 | This project implements a Model Context Protocol (MCP) server for Brave Search, allowing integration with AI assistants like Claude.
6 |
7 | ## Prerequisites
8 |
9 | - Python 3.11+
10 | - [uv](https://github.com/astral-sh/uv) - A fast Python package installer and resolver
11 |
12 | ## Installation
13 |
14 | ### Installing via Smithery
15 |
16 | To install Brave Search MCP server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@arben-adm/brave-mcp-search):
17 |
18 | ```bash
19 | npx -y @smithery/cli install @arben-adm/brave-mcp-search --client claude
20 | ```
21 |
22 | ### Manual Installation
23 | 1. Clone the repository:
24 | ```
25 | git clone https://github.com/your-username/brave-search-mcp.git
26 | cd brave-search-mcp
27 | ```
28 |
29 | 2. Create a virtual environment and install dependencies using uv:
30 | ```
31 | uv venv
32 | source .venv/bin/activate # On Windows, use: .venv\Scripts\activate
33 | uv pip install -r requirements.txt
34 | ```
35 |
36 | 3. Set up your Brave Search API key:
37 | ```
38 | export BRAVE_API_KEY=your_api_key_here
39 | ```
40 | On Windows, use: `set BRAVE_API_KEY=your_api_key_here`
41 |
42 | ## Usage
43 |
44 | 1. Configure your MCP settings file (e.g., `claude_desktop_config.json`) to include the Brave Search MCP server:
45 |
46 | ```json
47 | {
48 | "mcpServers": {
49 | "brave-search": {
50 | "command": "uv",
51 | "args": [
52 | "--directory",
53 | "path-to\\mcp-python\\brave-mcp-search\\src",
54 | "run",
55 | "server.py"
56 | ],
57 | "env": {
58 | "BRAVE_API_KEY": "YOUR_BRAVE_API_KEY_HERE"
59 | }
60 | }
61 | }
62 | }
63 | ```
64 |
65 | Replace `YOUR_BRAVE_API_KEY_HERE` with your actual Brave API key.
66 |
67 | 2. Start the Brave Search MCP server by running your MCP-compatible AI assistant with the updated configuration.
68 |
69 | 3. The server will now be running and ready to accept requests from MCP clients.
70 |
71 | 4. You can now use the Brave Search functionality in your MCP-compatible AI assistant (like Claude) by invoking the available tools.
72 |
73 | ## Available Tools
74 |
75 | The server provides two main tools:
76 |
77 | 1. `brave_web_search`: Performs a web search using the Brave Search API.
78 | 2. `brave_local_search`: Searches for local businesses and places.
79 |
80 | Refer to the tool docstrings in `src/server.py` for detailed usage information.
81 |
82 | ## Development
83 |
84 | To make changes to the project:
85 |
86 | 1. Modify the code in the `src` directory as needed.
87 | 2. Update the `requirements.txt` file if you add or remove dependencies:
88 | ```
89 | uv pip freeze > requirements.txt
90 | ```
91 | 3. Restart the server to apply changes.
92 |
93 | ## Troubleshooting
94 |
95 | If you encounter any issues:
96 |
97 | 1. Ensure your Brave API key is correctly set.
98 | 2. Check that all dependencies are installed.
99 | 3. Verify that you're using a compatible Python version.
100 | 4. If you make changes to the code, make sure to restart the server.
101 |
102 | ## License
103 |
104 | This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
105 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | mcp
2 | httpx
3 | fastmcp
4 |
5 | [dev]
6 | pytest
7 | black
8 | isort
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
1 | # Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
2 |
3 | startCommand:
4 | type: stdio
5 | configSchema:
6 | # JSON Schema defining the configuration options for the MCP.
7 | type: object
8 | required:
9 | - braveApiKey
10 | properties:
11 | braveApiKey:
12 | type: string
13 | description: The API key for the Brave Search server.
14 | commandFunction:
15 | # A function that produces the CLI command to start the MCP on stdio.
16 | |-
17 | (config) => ({command:'uv',args:['run', 'src/server.py'],env:{BRAVE_API_KEY:config.braveApiKey}})
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "brave-search-mcp"
3 | version = "0.1.0"
4 | description = "A Model Context Protocol (MCP) server for Brave Search"
5 | authors = [
6 | {name = "Your Name", email = "[email protected]"},
7 | ]
8 | dependencies = [
9 | "mcp",
10 | "httpx",
11 | "fastmcp",
12 | ]
13 | requires-python = ">=3.11"
14 | readme = "README.md"
15 | license = {text = "MIT"}
16 |
17 | [build-system]
18 | requires = ["setuptools>=61.0", "wheel"]
19 | build-backend = "setuptools.build_meta"
20 |
21 | [tool.setuptools]
22 | packages = ["src"]
23 |
24 | [project.optional-dependencies]
25 | dev = [
26 | "pytest",
27 | "black",
28 | "isort",
29 | ]
30 |
31 | [tool.black]
32 | line-length = 88
33 | target-version = ['py37']
34 |
35 | [tool.isort]
36 | profile = "black"
37 | line_length = 88
38 |
39 | [tool.pytest.ini_options]
40 | testpaths = ["tests"]
41 |
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | # Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
2 | # Use a Python image with uv pre-installed
3 | FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
4 |
5 | # Install the project into /app
6 | WORKDIR /app
7 |
8 | # Enable bytecode compilation
9 | ENV UV_COMPILE_BYTECODE=1
10 |
11 | # Copy from the cache instead of linking since it's a mounted volume
12 | ENV UV_LINK_MODE=copy
13 |
14 | # Install the project's dependencies using the lockfile and settings
15 | 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
16 |
17 | # Then, add the rest of the project source code and install it
18 | # Installing separately from its dependencies allows optimal layer caching
19 | ADD . /app
20 | RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev --no-editable
21 |
22 | FROM python:3.12-slim-bookworm
23 |
24 | WORKDIR /app
25 |
26 | COPY --from=uv /root/.local /root/.local
27 | COPY --from=uv --chown=app:app /app/.venv /app/.venv
28 |
29 | # Place executables in the environment at the front of the path
30 | ENV PATH="/app/.venv/bin:$PATH"
31 |
32 | # when running the container, add --db-path and a bind mount to the host's db file
33 | ENTRYPOINT ["uv", "run", "src/server.py"]
```
--------------------------------------------------------------------------------
/src/client.py:
--------------------------------------------------------------------------------
```python
1 | import asyncio
2 | import logging
3 | from mcp import ClientSession, StdioServerParameters
4 | from mcp.client.stdio import stdio_client
5 | from rich.console import Console
6 | from rich.logging import RichHandler
7 | from typing import Optional, Dict, Any
8 |
9 | logging.basicConfig(
10 | level=logging.INFO,
11 | format="%(message)s",
12 | datefmt="[%X]",
13 | handlers=[RichHandler()]
14 | )
15 |
16 | class BraveSearchClient:
17 | def __init__(
18 | self,
19 | server_path: str,
20 | api_key: str,
21 | console: Optional[Console] = None
22 | ):
23 | self.server_params = StdioServerParameters(
24 | command="python",
25 | args=[server_path],
26 | env={"BRAVE_API_KEY": api_key}
27 | )
28 | self.console = console or Console()
29 | self.logger = logging.getLogger("brave-search-client")
30 |
31 | def _is_complex_query(self, query: str) -> bool:
32 | """Determine if a query is complex based on its characteristics"""
33 | indicators = [
34 | " and ", " or ", " why ", " how ", " what ", " explain ",
35 | "compare", "difference", "analysis", "describe"
36 | ]
37 | return any(indicator in query.lower() for indicator in indicators) or len(query.split()) > 5
38 |
39 | async def _execute_search(
40 | self,
41 | session: ClientSession,
42 | tool: str,
43 | params: Dict[str, Any]
44 | ) -> str:
45 | try:
46 | # Adjust count based on query complexity
47 | if "query" in params:
48 | is_complex = self._is_complex_query(params["query"])
49 | params["count"] = 20 if is_complex else 10
50 |
51 | result = await session.call_tool(tool, params)
52 | if result.is_error:
53 | raise Exception(result.content[0].text)
54 | return result.content[0].text
55 | except Exception as e:
56 | self.logger.error(f"Search failed: {str(e)}")
57 | return f"Error: {str(e)}"
58 |
59 | async def run_interactive(self):
60 | """Run interactive search client"""
61 | try:
62 | async with stdio_client(self.server_params) as (read, write):
63 | async with ClientSession(read, write) as session:
64 | await session.initialize()
65 | tools = await session.list_tools()
66 |
67 | self.console.print(
68 | "Available tools:",
69 | ", ".join(tool.name for tool in tools)
70 | )
71 |
72 | while True:
73 | query = self.console.input("\nSearch query (or 'quit'): ")
74 | if query.lower() == "quit":
75 | break
76 |
77 | # Default to web search as it's used for answer formulation
78 | search_type = "web"
79 |
80 | is_complex = self._is_complex_query(query)
81 | count = 20 if is_complex else 10
82 |
83 | tool = "brave_web_search"
84 |
85 | with self.console.status(f"Searching with {'complex' if is_complex else 'standard'} query..."):
86 | result = await self._execute_search(
87 | session,
88 | tool,
89 | {"query": query, "count": count}
90 | )
91 |
92 | self.console.print("\nResults:", style="bold green")
93 | self.console.print(result)
94 |
95 | except Exception as e:
96 | self.logger.error(f"Client error: {str(e)}")
97 | raise
98 |
99 | if __name__ == "__main__":
100 | import os
101 | import sys
102 |
103 | if len(sys.argv) < 2:
104 | print("Usage: python client.py <path_to_server.py>")
105 | sys.exit(1)
106 |
107 | api_key = os.getenv("BRAVE_API_KEY")
108 | if not api_key:
109 | print("Error: BRAVE_API_KEY environment variable required")
110 | sys.exit(1)
111 |
112 | client = BraveSearchClient(sys.argv[1], api_key)
113 | asyncio.run(client.run_interactive())
```
--------------------------------------------------------------------------------
/src/server.py:
--------------------------------------------------------------------------------
```python
1 | from mcp.server.fastmcp import FastMCP
2 | import httpx
3 | import time
4 | import asyncio
5 | from typing import Optional, Dict, List, Any, Tuple
6 | from dataclasses import dataclass
7 | from enum import Enum
8 | import os
9 | import sys
10 | import io
11 |
12 | api_key = os.getenv("BRAVE_API_KEY")
13 | if not api_key:
14 | raise ValueError("BRAVE_API_KEY environment variable required")
15 |
16 | class RateLimitError(Exception):
17 | pass
18 |
19 | @dataclass
20 | class RateLimit:
21 | per_second: int = 1
22 | per_month: int = 2000
23 | _requests: Dict[str, int] = None
24 | _last_reset: float = 0.0
25 |
26 | def __post_init__(self):
27 | self._requests = {"second": 0, "month": 0}
28 | self._last_reset = time.time()
29 |
30 | def check(self):
31 | now = time.time()
32 | if now - self._last_reset > 1:
33 | self._requests["second"] = 0
34 | self._last_reset = now
35 |
36 | if (self._requests["second"] >= self.per_second or
37 | self._requests["month"] >= self.per_month):
38 | raise RateLimitError("Rate limit exceeded")
39 |
40 | self._requests["second"] += 1
41 | self._requests["month"] += 1
42 |
43 | class BraveSearchServer:
44 | def __init__(self, api_key: str):
45 | # Configure stdout for UTF-8
46 | if sys.platform == 'win32':
47 | sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
48 |
49 | self.mcp = FastMCP(
50 | "brave-search",
51 | dependencies=["httpx", "asyncio"]
52 | )
53 | self.api_key = api_key
54 | self.base_url = "https://api.search.brave.com/res/v1"
55 | self.rate_limit = RateLimit()
56 | self._client = None
57 | self._setup_tools()
58 |
59 | def get_client(self):
60 | if self._client is None or self._client.is_closed:
61 | self._client = httpx.AsyncClient(
62 | headers={
63 | "X-Subscription-Token": self.api_key,
64 | "Accept": "application/json",
65 | "Accept-Encoding": "gzip"
66 | },
67 | timeout=30.0
68 | )
69 | return self._client
70 |
71 | async def _get_web_results(self, query: str, min_results: int) -> List[Dict]:
72 | """Fetch web results with pagination until minimum count is reached"""
73 | client = self.get_client()
74 | self.rate_limit.check()
75 |
76 | try:
77 | # Make a single request with the maximum allowed count
78 | response = await client.get(
79 | f"{self.base_url}/web/search",
80 | params={
81 | "q": query,
82 | "count": min_results
83 | }
84 | )
85 | response.raise_for_status()
86 | data = response.json()
87 | results = data.get("web", {}).get("results", [])
88 | return results
89 | except httpx.HTTPStatusError as e:
90 | if e.response.status_code == 422:
91 | # If we get a 422, try with a smaller count
92 | response = await client.get(
93 | f"{self.base_url}/web/search",
94 | params={
95 | "q": query,
96 | "count": 10 # Fall back to smaller count
97 | }
98 | )
99 | response.raise_for_status()
100 | data = response.json()
101 | return data.get("web", {}).get("results", [])
102 | raise # Re-raise other HTTP errors
103 |
104 | def _format_web_results(self, data: Dict, min_results: int = 10) -> str:
105 | """Format web search results with enhanced information"""
106 | results = []
107 | web_results = data.get("web", {}).get("results", [])
108 |
109 | for result in web_results[:max(min_results, len(web_results))]:
110 | # Strip or replace any potential Unicode characters
111 | title = result.get('title', 'N/A').encode('ascii', 'replace').decode()
112 | desc = result.get('description', 'N/A').encode('ascii', 'replace').decode()
113 |
114 | formatted_result = [
115 | f"Title: {title}",
116 | f"Description: {desc}",
117 | f"URL: {result.get('url', 'N/A')}"
118 | ]
119 |
120 | # Add additional metadata if available
121 | if "meta_url" in result:
122 | formatted_result.append(f"Source: {result['meta_url']}")
123 | if "age" in result:
124 | formatted_result.append(f"Age: {result['age']}")
125 | if "language" in result:
126 | formatted_result.append(f"Language: {result['language']}")
127 |
128 | results.append("\n".join(formatted_result))
129 |
130 | return "\n\n".join(results)
131 |
132 | def _setup_tools(self):
133 | @self.mcp.tool()
134 | async def brave_web_search(
135 | query: str,
136 | count: Optional[int] = 20
137 | ) -> str:
138 | """Execute web search using Brave Search API with improved results
139 |
140 | Args:
141 | query: Search terms
142 | count: Desired number of results (10-20)
143 | """
144 | min_results = max(10, min(count, 20)) # Ensure between 10 and 20
145 |
146 | all_results = await self._get_web_results(query, min_results)
147 |
148 | if not all_results:
149 | return "No results found for the query."
150 |
151 | formatted_results = []
152 | for result in all_results[:min_results]:
153 | formatted_result = [
154 | f"Title: {result.get('title', 'N/A')}",
155 | f"Description: {result.get('description', 'N/A')}",
156 | f"URL: {result.get('url', 'N/A')}"
157 | ]
158 |
159 | # Include additional context if available
160 | if result.get('extra_snippets'):
161 | formatted_result.append("Additional Context:")
162 | formatted_result.extend([f"- {snippet}" for snippet in result['extra_snippets'][:2]])
163 |
164 | formatted_results.append("\n".join(formatted_result))
165 |
166 | return "\n\n".join(formatted_results)
167 |
168 | @self.mcp.tool()
169 | async def brave_local_search(
170 | query: str,
171 | count: Optional[int] = 20 # Changed default from 5 to 20
172 | ) -> str:
173 | """Search for local businesses and places
174 |
175 | Args:
176 | query: Location terms
177 | count: Results (1-20
178 | """
179 | self.rate_limit.check()
180 |
181 | # Initial location search
182 | params = {
183 | "q": query,
184 | "search_lang": "en",
185 | "result_filter": "locations",
186 | "count": 20 # Always request maximum results
187 | }
188 |
189 | client = self.get_client()
190 | response = await client.get(
191 | f"{self.base_url}/web/search",
192 | params=params
193 | )
194 | response.raise_for_status()
195 | data = response.json()
196 |
197 | location_ids = self._extract_location_ids(data)
198 | if not location_ids:
199 | # If no local results found, fallback to web search
200 | # with minimum 10 results
201 | return await brave_web_search(query, 20)
202 |
203 | # If we have less than 10 location IDs, try to get more
204 | offset = 0
205 | while len(location_ids) < 10 and offset < 40:
206 | offset += 20
207 | additional_response = await client.get(
208 | f"{self.base_url}/web/search",
209 | params={
210 | "q": query,
211 | "search_lang": "en",
212 | "result_filter": "locations",
213 | "count": 20,
214 | "offset": offset
215 | }
216 | )
217 | additional_data = additional_response.json()
218 | location_ids.extend(self._extract_location_ids(additional_data))
219 |
220 | # Get details for at least 10 locations
221 | pois, descriptions = await self._get_location_details(
222 | location_ids[:max(10, len(location_ids))]
223 | )
224 | return self._format_local_results(pois, descriptions)
225 |
226 | async def _get_location_details(
227 | self,
228 | ids: List[str]
229 | ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
230 | """Fetch POI and description data for locations"""
231 | client = self.get_client()
232 | pois_response, desc_response = await asyncio.gather(
233 | client.get(
234 | f"{self.base_url}/local/pois",
235 | params={"ids": ids}
236 | ),
237 | client.get(
238 | f"{self.base_url}/local/descriptions",
239 | params={"ids": ids}
240 | )
241 | )
242 | return (
243 | pois_response.json(),
244 | desc_response.json()
245 | )
246 |
247 | def _extract_location_ids(self, data: Dict) -> List[str]:
248 | """Extract location IDs from search response"""
249 | return [
250 | result["id"]
251 | for result in data.get("locations", {}).get("results", [])
252 | if "id" in result
253 | ]
254 |
255 | def _format_local_results(
256 | self,
257 | pois: Dict[str, Any],
258 | descriptions: Dict[str, Any]
259 | ) -> str:
260 | """Format local search results with details"""
261 | results = []
262 | for poi in pois.get("results", []):
263 | location = {
264 | "name": poi.get("name", "N/A"),
265 | "address": self._format_address(poi.get("address", {})),
266 | "phone": poi.get("phone", "N/A"),
267 | "rating": self._format_rating(poi.get("rating", {})),
268 | "price": poi.get("priceRange", "N/A"),
269 | "hours": ", ".join(poi.get("openingHours", [])) or "N/A",
270 | "description": descriptions.get("descriptions", {}).get(
271 | poi["id"], "No description available"
272 | )
273 | }
274 |
275 | results.append(
276 | f"Name: {location['name']}\n"
277 | f"Address: {location['address']}\n"
278 | f"Phone: {location['phone']}\n"
279 | f"Rating: {location['rating']}\n"
280 | f"Price Range: {location['price']}\n"
281 | f"Hours: {location['hours']}\n"
282 | f"Description: {location['description']}"
283 | )
284 |
285 | return "\n---\n".join(results) or "No local results found"
286 |
287 | def _format_address(self, addr: Dict) -> str:
288 | """Format address components"""
289 | components = [
290 | addr.get("streetAddress", ""),
291 | addr.get("addressLocality", ""),
292 | addr.get("addressRegion", ""),
293 | addr.get("postalCode", "")
294 | ]
295 | return ", ".join(filter(None, components)) or "N/A"
296 |
297 | def _format_rating(self, rating: Dict) -> str:
298 | """Format rating information"""
299 | if not rating:
300 | return "N/A"
301 | # Use ASCII star (*) instead of Unicode star
302 | stars = "*" * int(float(rating.get('ratingValue', 0)))
303 | return f"{rating.get('ratingValue', 'N/A')} {stars} ({rating.get('ratingCount', 0)} reviews)"
304 |
305 | def run(self):
306 | """Start the MCP server"""
307 | self.mcp.run()
308 |
309 | if __name__ == "__main__":
310 |
311 | server = BraveSearchServer(api_key)
312 | server.run()
313 |
```