# tavily_mcp

零依赖（只依赖 `requests`）的 Python 客户端，用于调用 [Tavily MCP HTTP 服务器](https://tavily.sharyuke.com/api/mcp)。可以在任何 Python 项目里直接调用 Tavily 的 `search` 工具 —— 不需要 MCP SDK，没有全局状态，没有框架绑定。

这个包是从 [lang2](../README.md) 这个"长线投资判断日报"项目里抽出来的；里面没有任何 lang2 专属的代码或配置。

## 安装

把 `tavily_mcp/` 整个目录丢进你的项目里就行（如果想做成正经的包，`pip install -e ./tavily_mcp`）。运行时只依赖一个包：

```bash
pip install requests
```

## 配置：两种方式选一种

### 方式 A：环境变量（最省事，推荐）

在 shell 或 `.env` 里设置一次：

```bash
export TAVILY_API_KEY="thb-xxxxx..."
# 以下都有默认值，可以不设
export TAVILY_MCP_URL="https://tavily.sharyuke.com/api/mcp"
export TAVILY_MCP_AUTH_HEADER="Authorization"
export TAVILY_MCP_TIMEOUT=60
export TAVILY_ENABLED=true
export TAVILY_MAX_RESULTS=20
export TAVILY_SEARCH_DEPTH=advanced   # basic | advanced
export TAVILY_TOPIC=news             # news | general
export TAVILY_DAYS=1
export TAVILY_INCLUDE_ANSWER=true
```

然后代码里直接用：

```python
from tavily_mcp import call_tavily_search

result = call_tavily_search("美联储利率决议最新进展")
```

识别的环境变量（全部可选）：`TAVILY_MCP_URL` / `TAVILY_API_KEY` / `TAVILY_MCP_AUTH_HEADER` / `TAVILY_MCP_TIMEOUT` / `TAVILY_ENABLED` / `TAVILY_MAX_RESULTS` / `TAVILY_SEARCH_DEPTH` / `TAVILY_TOPIC` / `TAVILY_DAYS` / `TAVILY_INCLUDE_ANSWER`。

### 方式 B：在 Python 里显式传入

```python
from tavily_mcp import TavilyClient, TavilyConfig

cfg = TavilyConfig(
    url="https://tavily.sharyuke.com/api/mcp",   # 默认就是这个
    auth_header_value="thb-xxxxx...",
    topic="news",
    days=1,
    max_results=20,
)

with TavilyClient(cfg) as client:
    result = client.search("AI chip earnings today")

print(result["answer"])   # AI 总结（如果 include_answer=True，默认就是）
for item in result["items"]:
    print(item["title"], "-", item["url"])
```

两种方式也可以混用 —— 显式 `TavilyConfig(...)` 优先级最高，没传字段的地方会从环境变量回退：

```python
# 只显式覆盖 URL，key 走环境变量
client = TavilyClient(TavilyConfig(url="https://my-proxy.example.com/mcp"))
```

## 每次调用都可以临时覆盖

`TavilyConfig` 里的所有字段都可以在单次 `search` 调用里临时覆盖，不用动 client：

```python
result = client.search(
    "OpenAI 融资消息",
    topic="general",
    days=7,
    max_results=5,
    include_domains=["reuters.com", "bloomberg.com"],
    search_depth="advanced",
    include_answer=False,
)
```

## 返回值结构

每次 `search`（包括模块级的便捷函数）返回的都是同一个 dict：

```python
{
    "items":    [{"title": str, "url": str, "snippet": str}, ...],
    "answer":   str | None,         # AI 总结（include_answer=True 时才有）
    "raw_text": str,                # 服务端返回的完整 markdown 块，调试用
    "error":    str | None,         # None 表示成功；否则是失败原因
}
```

**错误不会抛异常。** 任何传输、协议、服务器错误都会返回 `items=[]` + `error="..."`，调用方用 `if result["error"]` 检测就行。如果你想用异常风格，自己包一层：

```python
def search_or_raise(client, query, **kwargs):
    r = client.search(query, **kwargs)
    if r["error"]:
        raise RuntimeError(f"tavily: {r['error']}")
    return r
```

## 当前已实现的能力

- ✅ `search` 工具（完整参数：`query` / `topic` / `days` / `max_results` / `search_depth` / `include_answer` / `include_domains`）
- ✅ MCP `initialize` + `notifications/initialized` 握手，每个 client 实例自己缓存 session id
- ✅ HTTP 400/404 且 body 含 "session" 时自动重新握手
- 🔧 `extract` / `crawl` / `map` 三个工具还没实现，欢迎 PR —— 底层的 `_post` / `_ensure_session` 都已经准备好了

## 配置字段参考（`TavilyConfig`）

| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| `enabled` | `bool` | `True` | 设为 `False` 时 `search()` 立刻返回 `error="tavily disabled"`，不会真的发请求 |
| `url` | `str` | `https://tavily.sharyuke.com/api/mcp` | MCP 端点完整 URL |
| `auth_header_name` | `str` | `"Authorization"` | 鉴权 header 的字段名 |
| `auth_header_value` | `str` | `""` | 鉴权 header 的完整值（如 `"thb-xxxxx"`）。**留空会让 `search()` 直接返回 `error="no session"`** |
| `timeout_s` | `int` | `60` | 单次 HTTP 请求的超时秒数 |
| `max_results` | `int` | `10` | 转发给 `search.max_results` |
| `search_depth` | `"basic"` \| `"advanced"` | `"advanced"` | |
| `topic` | `"news"` \| `"general"` | `"news"` | |
| `days` | `int` | `1` | 时间窗提示，转发给 `search.days` |
| `include_answer` | `bool` | `True` | 是否同时请求 AI 总结段落 |

`TavilyConfig.safe_dict()` 会自动把 `auth_header_value` 替换成 `***redacted***`，适合直接打日志。

## 离线 smoke test（不需要 Tavily key）

```python
"""完全离线的解析器单元测试。"""
from tavily_mcp import parse_items, extract_answer

fake_text = """
## AI 总结
今天有几条值得关注的消息。
## 搜索结果 (3 条)
1. **第一条标题**
   URL: https://example.com/a
   这是第一行摘要。
   还有第二行。

2. **第二条标题**
   URL: https://example.com/b
   只有一行摘要。

3. **第三条标题**
   URL: https://example.com/c
   文末有点缀文字。
   Try unlimited access
"""

print("items:", parse_items(fake_text))
print("answer:", extract_answer(fake_text))
```

期望输出：

```
items: [{'title': '第一条标题', 'url': 'https://example.com/a', 'snippet': '这是第一行摘要。\n   还有第二行。'}, {'title': '第二条标题', 'url': 'https://example.com/b', 'snippet': '只有一行摘要。'}, {'title': '第三条标题', 'url': 'https://example.com/c', 'snippet': '文末有点缀文字。'}]
answer: 今天有几条值得关注的消息。
```

注意第三条的 `snippet` 已经把 `Try unlimited access` 这种推广尾巴剥掉了。

## 在其他项目里使用

最常见的两种用法：

```python
# 1. 一次性脚本：模块级便捷函数
from tavily_mcp import call_tavily_search
r = call_tavily_search("...query...")
for it in r["items"]:
    print(it["title"], it["url"])

# 2. 长跑服务：显式构造，复用 session
from tavily_mcp import TavilyClient, TavilyConfig
client = TavilyClient(TavilyConfig(auth_header_value="thb-..."))
# ... 业务循环 ...
for q in queries:
    r = client.search(q)
    handle(r)
client.close()  # 或者直接用 with 块
```

## 作为 kimi-code MCP server 运行（stdio）

这个包同时也实现了一个 stdio MCP server，可以直接被 [kimi-code](https://www.kimi.com/code/docs/en/kimi-code-cli/customization/mcp.html) 当成子进程拉起来，把 Tavily 暴露成本地工具给 agent 调用：

- 注册的工具名：`mcp__tavily__search` / `__extract` / `__crawl` / `__map`
- 协议：MCP 2024-11-05；framing 同时支持 **Content-Length framed JSON-RPC** 和 **newline-delimited JSON（NDJSON）**，自适应
- 网络请求全部透传到上游 `https://tavily.sharyuke.com/api/mcp`（不是本地实现）
- 没有新增第三方依赖，只用 `requests`

### 两种安装 / 注册方式

**A. 不安装，直接 `python -m tavily_mcp`**（改动最小，最快验证）

`~/.kimi-code/mcp.json`：

```json
{
  "mcpServers": {
    "tavily": {
      "command": "python",
      "args": ["-u", "-m", "tavily_mcp"],
      "cwd": "/Users/leaves/wineDir/kimi",
      "env": {
        "TAVILY_API_KEY": "thb-xxxxx",
        "TAVILY_MCP_DEBUG_LOG": "/tmp/tavily_mcp_debug.log"
      },
      "startupTimeoutMs": 45000,
      "toolTimeoutMs": 90000
    }
  }
}
```

`cwd` 指向 `tavily_mcp/` 的**上一级**（`python -u -m tavily_mcp` 在那一层能把
`tavily_mcp` 当 module 找到；用 `-u` 关掉 stdio block buffering）。
`TAVILY_MCP_DEBUG_LOG` 用来落地 debug 日志，方便排错（见末尾"排错 / 调试"）。

**B. 装到环境里得到 `tavily-mcp` 命令**（干净，但要 install 一次）

```bash
pip install -e /Users/leaves/wineDir/kimi/tavily_mcp
```

`~/.kimi-code/mcp.json`：

```json
{
  "mcpServers": {
    "tavily": {
      "command": "tavily-mcp",
      "env": { "TAVILY_API_KEY": "thb-xxxxx" }
    }
  }
}
```

### 手动验证（不需要 kimi-code）

把这个一次性脚本存成 `smoke.py` 跑一下，往 server 的 stdin 灌一个
最小 `initialize` + `tools/list` 请求：

```python
# smoke.py —— 一次性 MCP 烟雾测
import json, subprocess, sys, re

def frame(obj):
    body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
    return f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + body

reqs = [
    {"jsonrpc":"2.0","id":1,"method":"initialize",
     "params":{"protocolVersion":"2024-11-05","capabilities":{},
               "clientInfo":{"name":"smoke","version":"0"}}},
    {"jsonrpc":"2.0","method":"notifications/initialized"},
    {"jsonrpc":"2.0","id":2,"method":"tools/list"},
]
proc = subprocess.run(
    [sys.executable, "-u", "-m", "tavily_mcp"],
    cwd="/Users/leaves/wineDir/kimi",
    input=b"".join(frame(r) for r in reqs),
    capture_output=True,
)
print("stderr:", proc.stderr.decode().rstrip())
raw = proc.stdout
frames = [json.loads(raw[m.end():m.end()+int(m.group(1))].decode("utf-8"))
          for m in re.finditer(rb"Content-Length: (\d+)\r\n\r\n", raw)]
print(f"got {len(frames)} response frame(s):")
for f in frames:
    print(json.dumps(f, ensure_ascii=False, indent=2)[:400])
```

```bash
python3 smoke.py
```

期望：

- stderr 两行：`starting ...` 和 `stdin EOF; shutting down`
- stdout 第一帧 `id=1` 的 `result.serverInfo.name == "tavily-mcp"`，
  `protocolVersion == "2024-11-05"`
- stdout 第二帧 `id=2` 的 `result.tools` 长度 == 4，名字依次为
  `search` / `extract` / `crawl` / `map`
- `notifications/initialized` 不产生响应帧（通知语义）

### 工具语义

| 工具名 | 必填参数 | 透传给上游的字段 |
| --- | --- | --- |
| `search` | `query: string` | `topic`/`days`/`max_results`/`search_depth`/`include_answer`/`include_domains` 等 |
| `extract` | `urls: string[]` | `format` 等 |
| `crawl` | `url: string` | `max_depth`/`limit`/`instructions` 等 |
| `map` | `url: string` | `limit`/`depth` 等 |

所有 schema 都用 `additionalProperties: true`，agent 传任何字段都会被原样转发给上游 Tavily MCP，由上游做参数校验。

### 鉴权与环境变量

server 在启动时 `TavilyConfig.from_env()` 读 `TAVILY_*` 环境变量（详见上文"配置"节）。最常见的是把 `TAVILY_API_KEY` 通过 `mcp.json` 的 `env` 字段注进去；上游 url (`TAVILY_MCP_URL`)、深度、topic 等默认值也都可以通过同一机制覆盖。

### 排错 / 调试

每次 server 启动 / 每帧读 / 写都会同时打到 stderr 和
`TAVILY_MCP_DEBUG_LOG` 指向的文件（默认 `/tmp/tavily_mcp_debug.log`）。
内容带毫秒时间戳，类似：

```
[2026-08-10T10:48:31.870] [tavily_mcp:server] starting; python=3.14.6 pid=43302
[2026-08-10T10:48:31.871] [tavily_mcp:server] stdin <- 271 bytes: b'{"jsonrpc"...'
[2026-08-10T10:48:31.871] [tavily_mcp:server] frame accepted (ndjson, body=172B)
[2026-08-10T10:48:31.871] [tavily_mcp:server] stdout -> ndjson (161B)
```

如果 `/mcp` 显示 `failed · Timed out after Nms`，最常见的几条线索：

| debug log 看到 | 含义 |
| --- | --- |
| 只有 `starting ...`，后面空白 | stdin 没拿到任何字节 —— 检查 `cwd` / `args` / `env` |
| `stdin <- ...` 之后没 `frame accepted` | 输入 framing 不是 CL 也不是 NDJSON；按 `tail -f` 看具体字节再加解析 |
| `frame accepted` 后没 `stdout -> ...` | `_handle_*` 抛异常了；上一行会有 `handler crashed:` |
| `stdout -> ...` 但 kimi-code 仍 timeout | 是 stdout buffering；检查 args 里有没有 `-u` |

#### Framing 说明

`server.py` 同时支持两种 framing，第一条成功解析的 framing 会被记下，
之后响应也用同样的 framing 写回去（不混用）：

- **Content-Length**（MCP 标准）：`Content-Length: N\r\n\r\n<body>`
- **NDJSON**（更宽松）：一行一个 JSON 对象，以 `\n` 结尾

如果你的 kimi-code 版本不是用这两种之一，可以根据具体字节形
式在 `_try_frame()` 里加一个分支，再重测。

### 没验证的部分

- 上游 Tavily MCP HTTP 是否真的完整支持 `extract` / `crawl` / `map` —— 本地沙箱里没有真实 key 不能打。这条说明以 `tavily.sharyuke.com` 官方为准。
- 真实 key 下的端到端 `search` 调用需要你自己的 `thb-` key。