在 LangChain 里用 okdata：MCP 工具或一个小的 Python 封装

这一页最后会得到一个 Python Agent，能回答“这周 TikTok 上关于便携榨汁杯的视频都在讲什么，评论区主要吐槽什么”：它先搜 TikTok，再读点赞最高几条视频下的评论，最后总结。这类脚本适合增长分析同学每周一跑一次，也可以被工程师塞进更大的数据流程里。接法有两种：通过 MCP 服务把全部接口交给 Agent，或者只把需要的那一个接口包成普通工具。

准备

Python 3.10 或更新版本，外加一个 Agent 能用的对话模型。示例用的是 Anthropic 的模型，需要装 langchain-anthropic 并设置 ANTHROPIC_API_KEY；换成其他支持工具调用的模型也一样。
一个 okdata 的 key，以 okd_ 开头。新 key 自带 $0.10 额度。怎么拿 key 见快速开始。
把 key 放进环境变量 OKDATA_API_KEY。下面的代码只读取它，不会把它写进代码：

export OKDATA_API_KEY="okd_..."

方式一：通过 MCP 拿到全部接口

LangChain 自带的 MCP 支持在 langchain.mcp 里，需要 LangChain 1.4 及以上。它还标着 beta，运行时会提示一次 LangChainBetaWarning，不用管。

pip install "langchain[mcp]" langchain-anthropic

import asyncio
import os

from fastmcp.client import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

RULES = "每次收费调用 okdata 之前先说明费用，最多四次收费调用。"

async def main() -> None:
    client = Client("https://api.okdata.online/mcp", auth=os.environ["OKDATA_API_KEY"])  # 以 Authorization: Bearer <key> 发送
    async with MCPAdapter(client) as adapter:
        tools = await adapter.list_tools()  # find、describe、call、account
        agent = create_agent("claude-sonnet-5", tools, system_prompt=RULES)
        result = await agent.ainvoke({"messages": [{"role": "user", "content": (
            "在 TikTok 搜 portable blender，只要一页。读点赞最高的三条视频下的评论，"
            "列出主要吐槽点，附上链接。"
        )}]})
        print(result["messages"][-1].text)

asyncio.run(main())

Agent 会先用 find、describe 选接口、看价格，再用 call 取数据。

如果你的代码还在用旧的 langchain-mcp-adapters：它的 MultiServerMCPClient 照样能连，配置写 "transport": "streamable_http"、"url": "https://api.okdata.online/mcp"、"headers": {"Authorization": "Bearer " + os.environ["OKDATA_API_KEY"]} 即可。不过这个包已经不再维护，LangChain 官方有迁移到 langchain.mcp 的指南。

方式二：把一个接口写成工具

如果流程里永远只用一个接口，十几行 requests 代码比一条 MCP 连接更好读，也更好测。

pip install langchain langchain-anthropic requests

import os

import requests
from langchain.agents import create_agent
from langchain.tools import tool

API_URL = "https://api.okdata.online"

@tool
def tiktok_video_search(query: str) -> dict:
    """按关键词搜 TikTok 视频。一页约 30 条，带点赞、评论数和作者。
    每次调用 $0.002。"""
    response = requests.get(
        API_URL + "/v1/tiktok/videos/search",
        params={"query": query},
        headers={"Authorization": "Bearer " + os.environ["OKDATA_API_KEY"]},
        timeout=60,
    )
    # 出错时也会返回 JSON（code、message、retry_after_seconds），
    # 直接交给模型看，而不是抛异常。
    return response.json()

agent = create_agent("claude-sonnet-5", [tiktok_video_search],
                     system_prompt="每个问题最多搜两次。")
result = agent.invoke({"messages": [{"role": "user", "content": "这周 TikTok 上关于便携榨汁杯的视频都在拍什么？"}]})
print(result["messages"][-1].text)

所有接口都是同一个套路：https://api.okdata.online/v1/ 加上把接口 id 的点换成斜杠，参数放在查询字符串里。参数说明见各个接口页。

第一句可以这样说

在 TikTok 搜 "portable blender"，只要一页。读点赞最高的三条视频下的评论，
列出最常见的五个吐槽点，每个附一个链接。最后告诉我 okdata 一共花了多少。

花多少钱

find、describe、account 免费。上面这句话会调用一次 TikTok 视频搜索（$0.002），再读三条视频的评论（每条 $0.002），合计 {{cost:tiktok.videos.search1+tiktok.videos.comments.list3}}。小红书笔记搜索一页 $0.02。调用失败不扣费。每次返回的 meta.billing.amount 写着这次花了多少，顺手记进日志很方便。

出问题时

KeyError: 'OKDATA_API_KEY'。 运行 Python 的那个终端里没有设这个变量。export 之后再跑。
返回 unauthorized（401）。 key 送到了，但无效：打错了、少了字符，或者已经被吊销。
返回 insufficient_balance（402）。 余额不够这次调用，没有扣费。充值后重试，价格见定价页。
返回 rate_limited（429）。 一下子调用太多，并行跑多个 Agent 时很常见。错误里有 retry_after_seconds；在你自己的封装里 sleep 这么多秒，再发同一个请求就行，不扣费。
list_tools() 拿不到工具或连不上。 检查地址是不是以 /mcp 结尾，langchain 是否 1.4 及以上（pip show langchain），以及代码运行的机器能不能出网访问 https://api.okdata.online。

接着看

MCP 服务说明：四个工具做什么，以及该怎么叮嘱 Agent。
TikTok 视频搜索：参数、字段和价格。