pip install requests

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

from tavily_mcp import call_tavily_search

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

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"])

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

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

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

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

"""完全离线的解析器单元测试。"""
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: 今天有几条值得关注的消息。

# 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 块

{
  "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
    }
  }
}

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

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

# 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])

python3 smoke.py

[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)