Skip to content

用 MCP 给 AI 客户端加一个 IP 归属地查询工具(Python 实现,Claude/Cursor 都能接)

现在不少人把 Claude Desktop、Cursor 这类客户端当成工作台:丢一段 Nginx 日志过去让它分析,或者问「这批 IP 大概是哪个省的」。问题在于,模型自己上不了网,IP 段和城市对不上时它会顺着上下文猜,猜出来的地名看着挺像那么回事,查一下就是错的。

MCP(Model Context Protocol)解决的正是这件事:把「查 IP 归属地」做成一个工具挂在客户端上,模型需要时主动调用,拿到的是接口返回的真实数据。这篇从协议怎么走讲到完整代码,用 Python 写一个零第三方依赖的 MCP Server,接的是 IP9 的免费接口。

一、为什么这块适合做成 MCP 工具

先说清楚什么场景值得做。日常遇到的多是这几种:

  • 排障时把访问日志里的异常 IP 扔给模型,问「这些 IP 集中在哪几个地方」,判断是机器刷量还是真实用户;
  • 运营问「这周新增用户的 IP 都是哪儿的」,你想让模型直接给结论,而不是自己再去工具站一个个查;
  • 做地域化内容或者多语言站点,让模型按当前访客的 IP 归属地挑文案。

这三件事的共同点是:结论必须建立在真实查询结果上。模型没有联网能力,光靠训练数据里的片段记忆,把 IP 判别成哪个城市基本是碰运气。做成工具之后,模型只负责判断「该不该查、查哪个 IP、怎么解读结果」,数据归接口。

还有个小优势:MCP 是客户端侧的配置,不用改你自己的业务代码。写一个 server,配到客户端里就能用,同一份 server 也能被多个客户端复用。

二、MCP 的交互长什么样

不用把它想复杂,stdio 传输的 MCP 就是一行一个 JSON 的请求-响应。客户端往标准输入里写请求,你的程序往标准输出里写响应,就这么简单。

一次完整会话里,客户端会发这几种消息:

方法作用
initialize握手,互报协议版本和能力
notifications/initialized通知,不用回
tools/list拉取工具清单(名字、描述、参数 schema)
tools/call真正调用某个工具,带上参数
ping探活

响应统一是 {"jsonrpc":"2.0","id":<请求id>,"result":...}。注意两点:通知类消息没有 id,不能回响应;stdio 模式下每条消息必须在一行内,JSON 里不能有裸换行(json.dumps 默认就会转义,不用管)。

tools/list 里返回的 description 是给模型看的,写得含糊模型就不会选它。别写「查询 IP」,写清楚「根据 IPv4/IPv6 地址查询所属国家、省、市、运营商,用于判断访问来源地域」。这一步比代码本身更容易被忽略。

三、Python 实现

下面这份代码只用标准库(urllib + json + sys),不需要装任何包,一个文件就是完整的 MCP Server。暴露两个工具:lookup_ip 查指定 IP,lookup_my_ip 查调用方自己的 IP 归属地。

python
#!/usr/bin/env python3
# ip9_mcp_server.py —— 基于 IP9 免费接口的 IP 归属地 MCP Server(stdio 传输,零第三方依赖)
import json
import sys
import time
import urllib.parse
import urllib.request
from collections import OrderedDict

API = "https://ip9.com.cn/get"
QPS = 1.0            # 免费版 60 次/分钟,按 1 QPS 保守打
CACHE_TTL = 6 * 3600  # 归属地变化很慢,缓存 6 小时足够
CACHE_MAX = 20000

_cache = OrderedDict()      # key -> (过期时间戳, 结果字典)
_last_call = [0.0]          # 简单的令牌桶,只保留上一次调用时间


def _throttle():
    """限速:距离上次调用不足 1/QPS 秒就等一下。"""
    gap = 1.0 / QPS
    wait = _last_call[0] + gap - time.monotonic()
    if wait > 0:
        time.sleep(wait)
    _last_call[0] = time.monotonic()


def _fetch(ip=None):
    """查归属地。ip 为 None 时查请求方自身 IP。"""
    key = ip or "__self__"
    hit = _cache.get(key)
    if hit and hit[0] > time.time():
        _cache.move_to_end(key)
        return hit[1]

    url = API + ("?ip=" + urllib.parse.quote(ip) if ip else "")
    req = urllib.request.Request(url, headers={"User-Agent": "ip9-mcp/1.0"})

    _throttle()
    try:
        with urllib.request.urlopen(req, timeout=3) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except Exception as exc:                      # 网络异常、超时都不该让 server 挂掉
        return {"ok": False, "error": "请求失败:%s" % exc}

    if body.get("ret") == 400:
        return {"ok": False, "error": "IP 格式不合法(接口返回 ret=400),请确认是否为合法 IPv4/IPv6 地址"}
    if body.get("ret") == 429:
        time.sleep(2)
        return {"ok": False, "error": "触发免费版限速(ret=429),请稍后重试或降低调用频率"}
    data = body.get("data")
    if body.get("ret") != 200 or not data:
        return {"ok": False, "error": "接口返回异常:ret=%s" % body.get("ret")}

    result = {
        "ok": True,
        "ip": data.get("ip"),
        "国家": data.get("country"),
        "国家代码": data.get("country_code"),
        "省": data.get("prov"),
        "市": data.get("city"),
        "大区": data.get("big_area"),
        "运营商": data.get("isp"),
        "邮编": data.get("post_code"),
        "区号": data.get("area_code"),
        "经纬度": "%s,%s" % (data.get("lng"), data.get("lat")),
    }
    _cache[key] = (time.time() + CACHE_TTL, result)
    if len(_cache) > CACHE_MAX:
        _cache.popitem(last=False)                # 超量就淘汰最旧的
    return result


def _to_text(result):
    """把结果拼成模型好读的一段文本。"""
    if not result.get("ok"):
        return result.get("error", "查询失败")
    parts = [
        "IP:%s" % result["ip"],
        "归属:%s %s %s%s" % (result["国家"], result["省"], result["市"], result["大区"] or ""),
        "运营商:%s" % result["运营商"],
        "邮编/区号:%s / %s" % (result["邮编"], result["区号"]),
        "城市中心经纬度:%s" % result["经纬度"],
        "说明:经纬度为城市中心点,不是精确位置。",
    ]
    return "\n".join(parts)


TOOLS = [
    {
        "name": "lookup_ip",
        "description": "查询指定 IPv4/IPv6 地址的归属地,返回国家、省、市、大区、运营商、邮编区号。"
                       "分析访问日志、判断访问来源地域时使用。",
        "inputSchema": {
            "type": "object",
            "properties": {
                "ip": {"type": "string", "description": "要查询的 IP 地址,例如 58.246.100.1"}
            },
            "required": ["ip"],
        },
    },
    {
        "name": "lookup_my_ip",
        "description": "查询当前调用方自己的 IP 归属地(不传参数时接口返回请求方 IP),用于判断本机所在网络位置。",
        "inputSchema": {"type": "object", "properties": {}},
    },
]


def call_tool(name, args):
    if name == "lookup_ip":
        ip = (args or {}).get("ip", "").strip()
        if not ip:
            return "参数 ip 不能为空"
        return _to_text(_fetch(ip))
    if name == "lookup_my_ip":
        return _to_text(_fetch(None))
    return "未知工具:%s" % name


def handle(msg):
    """处理一条 JSON-RPC 消息,返回响应字典;通知类消息返回 None。"""
    method = msg.get("method")
    msg_id = msg.get("id")

    if method == "initialize":
        return {
            "jsonrpc": "2.0", "id": msg_id,
            "result": {
                "protocolVersion": msg.get("params", {}).get("protocolVersion", "2024-11-05"),
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "ip9-ip-lookup", "version": "1.0.0"},
            },
        }
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {"tools": TOOLS}}
    if method == "tools/call":
        params = msg.get("params") or {}
        text = call_tool(params.get("name"), params.get("arguments"))
        return {
            "jsonrpc": "2.0", "id": msg_id,
            "result": {"content": [{"type": "text", "text": text}], "isError": False},
        }
    if method == "ping":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {}}
    if msg_id is None:                 # 其他通知,直接忽略
        return None
    return {
        "jsonrpc": "2.0", "id": msg_id,
        "error": {"code": -32601, "message": "Method not found: %s" % method},
    }


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except ValueError:
            continue
        resp = handle(msg)
        if resp is None:
            continue
        sys.stdout.write(json.dumps(resp, ensure_ascii=False) + "\n")
        sys.stdout.flush()


if __name__ == "__main__":
    main()

存成 ip9_mcp_server.py,给它执行权限,然后先在终端里手测一下握手能不能通(这一步能省掉后面一大堆瞎猜):

bash
chmod +x ip9_mcp_server.py
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | ./ip9_mcp_server.py
echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"lookup_ip","arguments":{"ip":"58.246.100.1"}}}' | ./ip9_mcp_server.py

第二条命令会返回类似 上海 上海华东中国联通、邮编 200080 这样的文本——这就是模型最终看到的内容。

四、接到客户端里

以 Claude Desktop 为例,编辑配置文件(macOS 是 ~/Library/Application Support/Claude/claude_desktop_config.json,Windows 在 %APPDATA%\Claude\ 下),加一段:

json
{
  "mcpServers": {
    "ip9-ip-lookup": {
      "command": "python3",
      "args": ["/absolute/path/to/ip9_mcp_server.py"]
    }
  }
}

路径写绝对路径,command 写你环境里真实可用的解释器(用虚拟环境就写 venv 里的 python)。改完重启客户端,在工具列表里能看到 lookup_ip 就说明接通了。Cursor 的 mcp.json、其他支持 MCP 的客户端配置项名字略有差异,填的参数是同一套。

配好之后,直接把日志片段贴给模型,问「这些 IP 分别来自哪里,有没有集中在同一个机房」。模型会自己决定调几次 lookup_ip,再基于返回结果回答——这时候它给你的地名才是接口里的数据,不是编的。

五、几个需要注意的地方

工具描述决定模型会不会用对。 只写「IP 查询」,模型在分析日志时经常想不到调它;把「IPv4/IPv6」「归属地」「访问来源」这些词写进 description,命中率明显不一样。

不要一次把整段日志塞给模型。 日志有几千行的话,token 花在无关内容上,模型也容易挑错 IP。更省事的做法是让模型先看到「去重后的 IP 列表 + 出现次数」,或者干脆在 server 里加一个「批量统计」工具,把聚合做完再把 Top 结果交给模型。

缓存和限速必须做在 server 里。 免费版 60 次/分钟(约 1 秒 1 次),模型不会替你算额度,一轮对话连着查二三十个 IP 很正常。代码里 6 小时缓存加 1 QPS 令牌桶是最低配置;如果模型习惯一次问几十个 IP,把批量逻辑写进工具内部,比让模型逐个调用稳得多。

返回的经纬度别当精确位置用。 接口给的是城市中心坐标,同一个城市所有 IP 都是这一个点,用来在地图上打点可以,用来算两点距离就会离谱。

查到的是归属地,不是「人在哪儿」。 出差、代理、公司出口线路都会让 IP 归属地和实际位置不一致。让模型解读时加上这个前提,否则它很容易把「IP 显示上海」讲成「用户在上海」。

代码里调的就是 IP9 的免费接口:https://ip9.com.cn/get?ip=<IP>(不传参数则返回请求方自身 IP 的归属地,IPv4 和 IPv6 都支持,返回国家、省市、大区、运营商、邮编、区号、城市中心经纬度等字段),免注册、无需鉴权,免费版 60 次/分钟。接口细节可以在官网 https://www.ip9.com.cn 上核对。