brave-api

# 勇敢 API <img src="./brave-api.svg" alt="Brave API banner" /> <p align="center"> 一个用于 <a href="https://search.brave.com">Brave Search</a> 的异步 Python 客户端，提供流式 AI 答案和结构化的网页搜索，拥有内置的模型上下文协议（MCP）服务器。 </p> --- ## 目录 - [功能](#功能) - [架构](#架构) - [要求](#要求) - [安装](#安装) - [快速入门](#快速入门) - [提问](#提问) - [搜索](#搜索) - [客户端配置](#客户端配置) - [代理支持](#代理支持) - [对话](#对话) - [流式事件](#流式事件) - [流结果](#流结果) - [错误处理](#错误处理) - [MCP 服务器](#mcp-服务器) - [项目结构](#项目结构) - [示例](#示例) - [许可证](#许可证) --- ## 功能 **提问（AI）** - `client.ask()` — 阻塞调用，返回完整的 `StreamResult`，包含文本、信息框、图像、视频、网页结果和后续问题 - `client.ask_stream()` — 异步生成器，实时生成 `StreamEvent` 对象 - 通过 `conversation_id` 和 `symmetric_key` 支持多轮对话 - 多模态输入：附加图像到问题中（视觉） - 自动查询语言检测，手动覆盖 - 自动执行网页搜索、图像获取和其他工具调用 **搜索（网页）** - `client.search()` — 爬取结构化的网页和新闻结果，支持分页 - `client.suggest()` — 自动完成建议，支持实体检测 **MCP 服务器** - 将 `ask`、`search` 和 `suggest` 暴露为 MCP 工具，准备好集成到 Claude Desktop、Claude Code、Cursor 或任何兼容 MCP 的客户端 - 支持 **stdio**（本地客户端）和 **HTTP/SSE**（远程或多客户端部署）传输 - 完全通过环境变量配置，无需代码更改 - 与库共享相同的类型化客户端和错误层次结构 **常规** - 异步原生，基于 `curl_cffi`，带有浏览器指纹识别（无需 API 密钥） - 完整的 Pydantic 模型，用于运行时验证和类型安全 - 结构化的异常层次结构，用于可预测的错误处理 - 可配置的语言、国家、搜索安全、地理位置、超时和重试 - 可选的轮询代理池，自动直接连接回退 --- ## 架构 ```mermaid graph TD A[您的应用程序] -->|导入| B[BraveClient] C[MCP 客户端<br/>Claude Desktop / Claude Code / 其他] -->|stdio| D[Brave API MCP 服务器] G[OpenAI 兼容客户端<br/>或远程代理] -->|HTTP / SSE| D D -->|提问 / 搜索 / 建议| B B --> E[HTTP 客户端<br/>curl_cffi] E --> F[Brave Search / Brave AI] style B fill:#2b2b2b,stroke:#888,color:#fff style D fill:#2b2b2b,stroke:#888,color:#fff style F fill:#1a1a1a,stroke:#888,color:#fff ``` 该库可以直接在 Python 代码中使用，也可以通过 MCP 服务器间接使用，MCP 服务器包装了相同的 `BraveClient`，并将其暴露为 LLM 代理的工具。 --- ## 要求 - Python 3.11+ - 依赖项：`curl-cffi`、`pydantic`、`pillow` - 可选（MCP 服务器）：`fastmcp` --- ## 安装 ```bash uv pip install brave-api-python ``` 从源代码安装： ```bash git clone https://github.com/iqbalmh18/brave-api cd brave-api uv pip install -e ".[dev]" ``` 启用 MCP 服务器支持： ```bash uv pip install "brave-api-python[mcp]" ``` --- ## 快速入门 ```python import asyncio from brave_api import BraveClient async def main(): async with BraveClient() as client: # AI 答案 result = await client.ask("什么是量子计算？") print(result.text) # 网页搜索 search = await client.search("Python asyncio 教程") for item in search.web: print(item.title, item.url) asyncio.run(main()) ``` --- ## 提问 ### ask() — 阻塞，完整结果 ```python async with BraveClient() as client: result = await client.ask("印度尼西亚布罗莫山") print(result.text) # AI 答案文本（Markdown） if result.infobox: print(result.infobox.title) # "布罗莫山" print(result.infobox.subtitle) # "东爪哇的活火山" print(result.infobox.url) # 维基百科网址 print(result.infobox.image_url) # 实体图像 for img in result.images: print(img.url, img.thumbnail) for vid in result.videos: print(vid.title, vid.url) for web in result.web_results: print(web.title, web.url) for q in result.followups: print(q) ``` 带有图像（视觉/多模态）： ```python from pathlib import Path result = await client.ask("这张图像是什么？", image=Path("photo.jpg")) ``` ### ask_stream() — 实时流式 ```python async for event in client.ask_stream("Space X 是什么？"): if event.type is StreamEventType.TEXT_DELTA: print(event.delta, end="", flush=True) elif event.type is StreamEventType.TEXT_STOP: print() elif event.type is StreamEventType.FOLLOWUPS: for q in event.payload.get("followups", []): print(f"-> {q}") ``` ### 比较 | 方法 | 模式 | 返回值 | 最适合 | |---|---|---|---| | `client.ask()` | 阻塞 | `StreamResult` | 一次获取完整结果（信息框、图像等） | | `client.ask_stream()` | 流式 | `AsyncGenerator[StreamEvent]` | 打字机式输出 | | `conversation()` + `collect()` | 阻塞 | `StreamResult` | 多轮、图像输入、完全控制 | | `conversation()` + `stream_events()` | 流式 | `AsyncGenerator[StreamEvent]` | 流式 + 多轮 | --- ## 搜索 ### search() — 网页和新闻结果 ```python async with BraveClient() as client: result = await client.search("Python asyncio 教程") print(result.query) # 原始查询 print(len(result.web)) # 网页结果数量 print(len(result.news)) # 新闻结果数量 for item in result.web: print(item.title) print(item.url) print(item.description) print(item.age) # "2 天前" 等 for item in result.news: print(item.title, item.source, item.age) # 所有唯一网址的扁平列表 for url in result.urls: print(url) ``` 分页： ```python # 第 1 页（默认） page1 = await client.search("Rust 编程", offset=0) # 第 2 页 page2 = await client.search("Rust 编程", offset=1) ``` 禁用拼写检查以进行精确关键词匹配： ```python result = await client.search("pyton 教程", spellcheck=False) ``` ### suggest() — 自动完成 ```python suggestions = await client.suggest("elon") for s in suggestions: print(s.text, s.entity_type, s.is_entity) if s.thumbnail: print(s.thumbnail) ``` --- ## 客户端配置 `ClientConfig` 是一个冻结的 Pydantic 模型。所有字段都有安全的默认值。 ```python from brave_api import BraveClient, ClientConfig config = ClientConfig( # 语言和地区 language="id", # 响应语言："id"、"en" 等 ui_lang="id-id", # UI 语言："id-id"、"en-us" 等 country="id", # ISO 3166-1 国家代码 geoloc="-6.200x106.816", # 纬度 x 经度（雅加达） # 搜索 safesearch="moderate", # "off"、"moderate" 或 "strict" units_of_measurement="metric", # "metric" 或 "imperial" # 模式 enable_research=False, # True = 深度研究模式 # HTTP request_timeout_seconds=60.0, max_retries=3, retry_backoff_seconds=1.5, # 可选代理池 proxy_list=[ "http://user:password@proxy-1.example:8080", "socks5://proxy-2.example:1080", ], # 浏览器指纹识别 impersonate="chrome136", extra_headers={"X-Custom": "value"}, ) async with BraveClient(config) as client: ... ``` --- ## 代理支持 通过 `ClientConfig(proxy_list=...)` 传递代理 URL 以轮换请求中的代理使用。代理池由 Ask、Search、Suggest 和流式请求共享。 ```python from brave_api import BraveClient, ClientConfig config = ClientConfig( proxy_list=[ "http://user:password@proxy-1.example:8080", "http://proxy-2.example:8080", "socks5://proxy-3.example:1080", ], ) async with BraveClient(config) as client: result = await client.search("Python asyncio 教程") ``` 支持的代理方案包括 `http`、`https`、`socks4`、`socks4a`、`socks5` 和 `socks5h`。代理 URL 在创建配置时会被规范化和重复项会被删除。 客户端使用循环顺序为每个请求选择一个活动的代理。如果代理在收到响应或流式数据之前失败，则该代理将被禁用，直到客户端的生命周期结束。会自动尝试下一个活动的代理；如果没有剩余的代理，则请求将继续使用 `curl_cffi` 配置的直接连接（`proxies=None`）。来自 Brave 的 HTTP 响应，包括 `429` 和 `5xx`，不会禁用代理。 启用传输日志以查看所选代理和回退行为。代理凭证永远不会包含在这些日志消息中。 ```python import logging logging.basicConfig( level=logging.DEBUG, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) ``` 参见 [`examples/proxy-usage.py`](examples/proxy-usage.py) 以获取一个可运行的示例，该示例从 `BRAVE_PROXY_LIST` 读取以逗号分隔的代理 URL。 --- ## 对话 ```python async with BraveClient() as client: # 新对话 conv = await client.conversation("解释 DNS 的工作原理") result = await conv.collect() # 继续相同的对话 conv2 = await client.conversation( "什么是 DNSSEC？", conversation_id=conv.id, symmetric_key=conv.symmetric_key, ) result2 = await conv2.collect() ``` 关键 `conversation()` 参数： | 参数 | 类型 | 描述 | |---|---|---| | `query` | `str` | 问题或提示（必填） | | `conversation_id` | `str \| None` | 继续现有的对话 | | `symmetric_key` | `str \| None` | 在设置 `conversation_id` 时必填 | | `image` | `bytes \| str \| Path \| None` | 多模态输入的图像 | | `language` | `str \| None` | 重写响应语言 | | `query_type` | `str` | 参见 `QueryType` 枚举 | | `auto_tools` | `bool` | 自动执行工具调用（默认值：`True`） | | `context` | `str \| None` | 文章/段落上下文 | | `quote` | `str \| None` | 高亮文本跨度 | --- ## 流式事件 ```python async for event in conv.stream_events(): if event.type is StreamEventType.TEXT_DELTA: print(event.delta, end="", flush=True) elif event.type is StreamEventType.TEXT_STOP: print() elif event.type is StreamEventType.ERROR: print(f"错误：{event.error_message}") ``` 关键事件类型： ``` TEXT_DELTA / TEXT_STOP 响应文本令牌 THINKING_DELTA / THINKING_STOP 思维链推理 TOOL_USE 服务器请求工具调用 AUGMENT_WITH_TOOL_USE 运行工具结果（网页结果、图像等） AUGMENT_WITH_WEB / NEWS / IMAGES / VIDEOS 丰富数据 AUGMENT_WITH_INFOBOX 实体知识卡 FOLLOWUPS 建议后续问题 ERROR 服务器错误事件 CHALLENGE 需要 CAPTCHA ``` 下面的序列显示了这些事件在单个 `ask()` 调用期间如何流动： ```mermaid sequenceDiagram participant App participant Client as BraveClient participant Brave as Brave AI App->>Client: ask(query) Client->>Brave: POST /api/tap/v1/new Brave-->>Client: SSE 流 loop 直到流结束 Brave-->>Client: TEXT_DELTA / THINKING_DELTA Brave-->>Client: AUGMENT_WITH_WEB / IMAGES / INFOBOX end Brave-->>Client: FOLLOWUPS Client-->>App: StreamResult ``` --- ## 流结果 ```python result = await conv.collect() result.text # str - 完整 AI 答案（Markdown） result.thinking # str - 思维链推理（如果有） result.urls # list[str] - 找到的唯一 URL result.images # list[ImageResult] result.videos # list[VideoResult] result.web_results # list[WebResult] result.infobox # Infobox | None result.followups # list[str] result.citations # list[dict] - 原始工具结果有效载体 result.inline_entities # list[dict] result.raw_events # list[StreamEvent] - 每个事件用于调试 result.state # StreamState 枚举 result.is_complete # bool result.has_images # bool result.has_videos # bool result.has_infobox # bool result.has_tool_calls # bool ``` --- ## 错误处理 所有异常都继承自 `BraveAPIError`。 ``` BraveAPIError ├── TransportError 网络错误、超时、连接重置 ├── HTTPStatusError 非 2xx HTTP 响应（.status_code, .response_text） ├── TokenExtractionError 无法从服务器 HTML 中解析身份验证令牌 ├── ConversationError /api/tap/v1/new 未返回对话 ID ├── StreamAbortedError 服务器在中间流式发送错误 ├── ChallengeRequiredError 需要 CAPTCHA ├── InvalidResponseError 响应不是有效的 JSON 或意外形状 ``` ```python from brave_api.exceptions import ( BraveAPIError, ChallengeRequiredError, HTTPStatusError, StreamAbortedError, TransportError, ) try: async with BraveClient() as client: result = await client.ask("Rust 是什么？") except ChallengeRequiredError: print("需要 CAPTCHA") except HTTPStatusError as e: print(f"HTTP {e.status_code}: {e.response_text[:200]}") except TransportError as e: print(f"网络错误：{e}") except StreamAbortedError as e: print(f"流中止：{e}") except BraveAPIError as e: print(f"错误：{e}") ``` 重试策略：HTTP 429 和 5xx 响应会在出现指数退避（`backoff_seconds * 2^attempt`）时重试。 MCP 服务器重用此相同的层次结构：客户端引发的任何 `BraveAPIError` 都会被捕获，并作为 `ToolError` 以干净、描述性的消息传递给调用的 MCP 客户端，而不是原始的堆栈跟踪。 --- ## MCP 服务器 <img src="./brave-api-mcp.svg" alt="Brave API MCP banner" /> Brave API 配备了一个 [模型上下文协议](https://modelcontextprotocol.io) 服务器，基于 [FastMCP](https://gofastmcp.com) 构建，将客户端的核心功能暴露为任何兼容 MCP 的代理（Claude Desktop、Claude Code、Cursor 等）的工具。 ### 工具 | 工具 | 描述 | 只读 | |---|---|---| | `ask` | 向 Brave AI 提问并接收带有引文、来源网址、图像、视频和后续建议的完整 AI 生成答案 | 是 | | `search` | 搜索 Brave 并返回结构化的网页和新闻结果（原始 SERP，无 AI 答案） | 是 | | `suggest` | 为部分查询获取自动完成建议，包括带有缩略图的丰富实体建议 | 是 | ### 运行服务器 **stdio**（默认 - 用于本地客户端，如 Claude Desktop、Claude Code、Cursor）： ```bash python -m brave_api.mcp.server # 或通过 CLI 入口点 brave-api-mcp ``` 服务器通过 stdio 进行通信，旨在由 MCP 客户端启动，而不是在终端中独立运行以进行交互式使用。 **HTTP**（用于远程或多客户端部署、OpenAI 兼容客户端）： ```bash brave-api-mcp --http # 绑定到特定主机/端口 brave-api-mcp --http --host 0.0.0.0 --port 8000 ``` 完整的 CLI 参考： ``` 用法：brave-api-mcp [-h] [--http] [--host HOST] [--port PORT] [--log-level {debug,info,warning,error,critical}] 选项： --http 使用 HTTP/SSE 传输代替 stdio。 --host HOST 绑定 HTTP 传输的主机地址（仅限）。[默认值：127.0.0.1] --port PORT 绑定 HTTP 传输的端口（仅限）。[默认值：8000] --log-level LEVEL 日志级别。[默认值：warning] ``` ### 配置 MCP 客户端 **Claude Desktop / Claude Code** (`claude_desktop_config.json` 或 `~/.claude/claude_code_config.json`）： ```json { "mcpServers": { "brave-api": { "command": "python", "args": ["-m", "brave_api.mcp.server"], "env": { "BRAVE_COUNTRY": "id", "BRAVE_LANGUAGE": "id", "BRAVE_SAFESEARCH": "moderate" } } } } ``` 或通过 Claude Code CLI： ```bash claude mcp add brave-api python -- -m brave_api.mcp.server ``` **OpenAI 兼容客户端/远程部署** - 以 HTTP 模式启动服务器，并将客户端指向端点： ```bash brave-api-mcp --http --host 0.0.0.0 --port 8000 ``` 服务器在 `http://<host>:<port>/mcp` 处暴露一个标准的 MCP-over-HTTP（Streamable HTTP / SSE）端点。任何支持 MCP HTTP 传输的客户端都可以直接连接到它。 ### 环境变量 所有服务器行为均通过环境变量控制 - 无需更改代码。 | 变量 | 默认值 | 描述 | |---|---|---| | `BRAVE_BASE_URL` | `https://search.brave.com` | Brave 端点的基准 URL | | `BRAVE_GEOLOC` | 库默认值 | 地理位置为 `lat x lng` | | `BRAVE_COUNTRY` | 库默认值 | ISO 3166-1 国家代码 | | `BRAVE_LANGUAGE` | 库默认值 | 响应语言（BCP-47） | | `BRAVE_UI_LANG` | 库默认值 | UI 语言，例如 `en-us` | | `BRAVE_SAFESEARCH` | 库默认值 | `off`、`moderate` 或 `strict` | | `BRAVE_ENABLE_RESEARCH` | `false` | `true`/`false`/`1`/`0`/`yes`/`no` - 启用深度研究模式 | | `BRAVE_REQUEST_TIMEOUT` | 库默认值 | 请求超时（秒） | | `BRAVE_MAX_RETRIES` | 库默认值 | 最大重试次数 | | `BRAVE_MAX_CONCURRENT` | 库默认值 | 最大并发请求数 | 无效的数字或布尔值会回退到其默认值，并在启动时记录警告而不是引发错误。 ### 错误面 每个工具调用都会包装，以便底层客户端引发的任何 `BraveAPIError` 都转换为 MCP `ToolError`，并带有原始消息，保持直接库使用和基于 MCP 的使用之间的错误处理一致性。 --- ## 示例 | 文件 | 描述 | |---|---| | `examples/quick_start.py` | 最简单的用法 - 提问并打印 | | `examples/stream_events.py` | 实时令牌流式传输 | | `examples/web_results_and_urls.py` | 网页结果、缩略图、网址 | | `examples/images_and_videos.py` | 图像和视频结果 | | `examples/multi_turn_conversation.py` | 多轮 + 答案再生 | | `examples/client_config.py` | 所有 ClientConfig 选项 | | `examples/language_override.py` | 自动检测 vs 显式语言 | | `examples/multimodal_image_input.py` | 视觉 - 将图像附加到查询 | | `examples/context_and_quote.py` | 上下文和引用参数 | | `examples/auto_tools_control.py` | auto_tools=True vs False | | `examples/exception_handling.py` | 所有异常类型 + 流状态 | | `examples/inline_entities_and_citations.py` | 内联实体和工具引用 | | `examples/context_manager_vs_manual.py` | 客户端生命周期模式 | | `examples/raw_events_inspection.py` | 检查每个原始事件 | | `examples/thinking_mode.py` | 思维链推理 | | `examples/interactive_chat.py` | 终端 REPL 聊天 | | `examples/ask_method.py` | ask() 和 ask_stream() 演示 | | `examples/search_method.py` | search() 和 suggest() | | `examples/proxy-usage.py` | 代理池轮换、回退和传输日志 | --- ## 许可证 该项目根据 [LICENSE](https://github.com/iqbalmh18/brave-api/blob/main/LICENSE) 文件中的许可条款获得许可。