“你将学到设计一个企业级日志 MCP 的工具分层Discovery 层 Query 层让 AI 不再蒙头乱查用 FastMCP 把 Loki 或 Elasticsearch 的查询接口封装成 MCP 工具完整 Python 代码选对传输协议内网 stdio 还是 HTTP安全性和复杂度如何取舍在 MCP 中处理私有认证Bearer token / API key / Basic auth且不暴露明文凭证一、AI 跟你一起对着日志干瞪眼我见过太多这样的场景。生产告警在半夜响了on-call 工程师打开 Claude Code把报错日志粘进去问这是什么原因AI 给了一个看起来很有道理的分析但这个分析基于的是那几十行日志片段——而真正的根因藏在上游服务三分钟前的一条 ERROR藏在某个 pod 的完整调用链里藏在你没有粘贴进去的地方。AI 的分析不是错的只是不完整。因为它只能看到你告诉它的那部分。这就是问题的本质AI 看日志不是 AI 没能力分析日志而是 AI 根本没有主动查日志的渠道。你在手动做信息中间人的工作——从 Kibana 或 Grafana 复制日志粘到 Claude看 Claude 分析再去系统里找更多上下文再粘循环。这一讲要解决的就是这个信息中间人问题。把企业日志系统包装成 MCP让 Claude Code 能直接调用查询接口从看你粘贴的片段升级到自己去查、自己判断要看什么。这个痛点的根源清楚了。我们先搞懂为什么这件事不简单然后再一步步把它做出来。二、为什么不能直接用官方 MCP要自己写这个问题值得先回答因为很多人第一反应是去找现成的。Grafana 官方确实有mcp-grafana仓库Elastic 也有mcp-server-elasticsearch当前 v0.4.6。如果你的日志系统是标准的公有云 Loki 或者 Elastic Cloud这些官方实现装上去就能用。但企业内网的日志系统通常有三个官方 MCP 处理不了的问题第一工具粒度太粗。官方 Loki MCP 只暴露一个loki_query工具要求 AI 直接输入 LogQL 查询语句。问题是如果 AI 不知道你的系统里有哪些 labelapp、env、namespace……它只能靠猜。猜出来的 LogQL 十有八九查不到东西而且你不知道是没日志还是 label 写错了。正确的做法是先 Discovery探索有什么再 Query有目的地查——这需要多个分层工具。第二认证方式不匹配。企业内网的日志系统认证千变万化有用 LDAP 打通单点登录的有每个团队发 API key 的有走内部 OAuth 的还有用 Cookie 维持会话的。官方 MCP 支持的认证方式往往是针对公有云设计的对内网私有部署支持有限。第三查询权限不可控。把一个能全量查询生产日志的 MCP 工具直接暴露给 AI等于把你们的 DBA 账号直接给了 AI 用。你需要在 MCP 层做最小权限控制只允许查特定时间范围、特定服务、特定字段禁止写入、禁止删除索引。这三个问题加在一起解释了为什么企业场景下自定义 MCP 比用官方实现更合适。官方 vs 自定义 MCP 工具设计这三个问题搞清楚之后我们来设计解决方案。三、企业日志 MCP 的架构设计在动手写代码之前先把架构想清楚。一个能在生产环境用的日志 MCP需要回答三个设计问题。工具要怎么分层我在这个项目里采用了两层设计Discovery 层和Query 层。Discovery 层的工具负责告诉 AI 这个日志系统里有什么——有哪些 label、哪些 label 的可能值、哪些服务正在产生日志。这层工具是无害的调用它们不会产生大量数据传输不会有性能影响。Query 层的工具负责真正去查日志——按时间范围查日志流、在特定时间点执行表达式。这层工具有潜在的性能影响需要做查询限制比如返回条数上限、时间范围上限。这个分层设计让 AI 的查询行为更接近有经验的工程师先探索再精确查询而不是上来就发一个全量扫描把服务器打挂。Loki 的工具设计层级工具名功能对应 Loki APIDiscoverylist_labels列出所有可用 label/loki/api/v1/labelsDiscoverylist_label_values列出某个 label 的所有值/loki/api/v1/label/{name}/valuesDiscoverylist_streams列出活跃的日志流/loki/api/v1/seriesQueryquery_logs按时间范围查日志/loki/api/v1/query_rangeQueryinstant_query在特定时间点执行 LogQL/loki/api/v1/query传输协议选哪个MCP 支持两种传输协议stdio标准输入输出和Streamable HTTP。对企业内网部署stdio 是首选。原因很实际stdio 模式下MCP server 是 Claude Code 的子进程认证凭证通过环境变量传入不需要网络暴露不需要搭 HTTP 服务器没有端口占用没有额外的网络攻击面Claude Code 的claude_desktop_config.json或settings.json直接配置 env凭证管理清晰什么时候用 Streamable HTTP当你需要多个工程师共享同一个 MCP server 实例或者 MCP server 需要部署在独立机器上比如有网络隔离的堡垒机时。这种情况下才需要上 OAuth 2.1复杂度高得多第 33 讲会专门讲数据库 MCP 的 HTTP 部署方案。stdio vs HTTP 传输协议部署架构认证方案内网 Loki 常见认证Basic auth用户名 密码最常见通过Authorization: Basic base64头传递Bearer token通过Authorization: Bearer token头传递多租户Grafana Loki额外需要X-Scope-OrgID头内网 Elasticsearch 常见认证API key通过Authorization: ApiKey key头传递Basic auth同 Loki不少企业部署了内部 Elastic 时直接关掉了 security不推荐但确实存在这些认证信息全部通过环境变量注入 stdio 进程不硬编码在代码里不提交到 git。Claude Code 通过 MCP 查询日志完整流程架构说清楚了我们开始写代码。在这里大家需要学会用 AI 自己创建 MCP 给自己使用。你完全可以结合码哥之前分享的使用 /superpower 来头脑风暴把上文提出的 MCP 架构和实现方式告诉 Claude Code并且在提示语中提示可以使用 /mcp-builder 创建最终的 skill。四、实战把 Loki 查询包装成 MCP完整代码4.1 环境准备Python 3.10 mcp[cli] 1.27.22026-05-29 发布截至本讲写作时的最新版 httpx 0.28.x异步 HTTP 客户端安装依赖pip install mcp[cli] httpx“版本提醒MCP Python SDK 迭代快如果你在几个月后读到这篇先用pip show mcp确认版本mcp.tool()这个装饰器 API 在 1.x 大版本内是稳定的。4.2 第一步创建 MCP server 骨架其实大家在创建 MCP 的时候可以结合 superpower 做头脑风暴并且提示使用 mcp-builder skill 看来创建 MCP。而不是全部手写代码实现。新建文件loki_mcp/server.py# loki_mcp/server.py import os import base64 from datetime import datetime, timedelta, timezone import httpx from mcp.server.fastmcp import FastMCP # --- 初始化 FastMCP server --- mcp FastMCP(loki-log-server) # --- 从环境变量读取认证配置 --- LOKI_URL os.environ.get(LOKI_URL, http://localhost:3100) LOKI_USERNAME os.environ.get(LOKI_USERNAME, ) LOKI_PASSWORD os.environ.get(LOKI_PASSWORD, ) LOKI_BEARER_TOKEN os.environ.get(LOKI_BEARER_TOKEN, ) LOKI_ORG_ID os.environ.get(LOKI_ORG_ID, ) # Grafana 多租户头 # --- 查询安全限制 --- MAX_LIMIT int(os.environ.get(LOKI_MAX_LIMIT, 1000)) # 单次最多返回条数 MAX_HOURS int(os.environ.get(LOKI_MAX_HOURS, 24)) # 最多回溯小时数 def _build_headers() - dict[str, str]: 根据环境变量构建认证 headers优先级Bearer Basic 无认证 headers {Content-Type: application/json} if LOKI_BEARER_TOKEN: headers[Authorization] fBearer {LOKI_BEARER_TOKEN} elif LOKI_USERNAME and LOKI_PASSWORD: credentials base64.b64encode( f{LOKI_USERNAME}:{LOKI_PASSWORD}.encode() ).decode() headers[Authorization] fBasic {credentials} if LOKI_ORG_ID: headers[X-Scope-OrgID] LOKI_ORG_ID return headers def _validate_time_range(start_hours: float, end_hours: float) - tuple[str, str]: 把N 小时前转成 Loki 需要的 nanosecond 时间戳字符串。 同时做安全校验时间范围不超过 MAX_HOURS。 if start_hours 0or end_hours 0: raise ValueError(时间范围不能是负数) if start_hours MAX_HOURS: raise ValueError(f最多只能查 {MAX_HOURS} 小时内的日志当前请求 {start_hours} 小时) now datetime.now(timezone.utc) start_dt now - timedelta(hoursstart_hours) end_dt now - timedelta(hoursend_hours) # Loki API 使用 nanosecond 时间戳 start_ns str(int(start_dt.timestamp() * 1e9)) end_ns str(int(end_dt.timestamp() * 1e9)) return start_ns, end_ns4.3 第二步实现 Discovery 层工具继续在server.py中添加 Discovery 层工具mcp.tool() asyncdef list_labels() - dict: 列出 Loki 中所有可用的 label 名称。 在构造查询之前先用这个工具了解当前日志系统的结构。 返回示例{labels: [app, env, namespace, pod, level]} asyncwith httpx.AsyncClient(timeout10.0) as client: resp await client.get( f{LOKI_URL}/loki/api/v1/labels, headers_build_headers(), verifyFalse# 内网自签名证书常见生产环境替换为 CA 路径 ) resp.raise_for_status() data resp.json() return {labels: data.get(data, [])} mcp.tool() asyncdef list_label_values(label_name: str) - dict: 列出指定 label 的所有可能值。 例如list_label_values(app) 返回 [order-service, payment-service, gateway] Args: label_name: label 名称必须是 list_labels 返回的 label 之一 ifnot label_name ornot label_name.replace(-, ).replace(_, ).isalnum(): raise ValueError(f无效的 label 名称: {label_name!r}) asyncwith httpx.AsyncClient(timeout10.0) as client: resp await client.get( f{LOKI_URL}/loki/api/v1/label/{label_name}/values, headers_build_headers(), verifyFalse ) resp.raise_for_status() data resp.json() return { label: label_name, values: data.get(data, []) } mcp.tool() asyncdef list_streams( match: str , start_hours: float 1.0 ) - dict: 列出当前活跃的日志流流 一组 label 的组合。 Args: match: LogQL stream selector例如 {apporder-service} 空字符串返回所有流数量可能很大建议加过滤 start_hours: 往前看多少小时默认 1 小时 start_ns, end_ns _validate_time_range(start_hours, 0) params: dict {start: start_ns, end: end_ns, limit: 100} if match: params[match[]] match asyncwith httpx.AsyncClient(timeout15.0) as client: resp await client.get( f{LOKI_URL}/loki/api/v1/series, headers_build_headers(), paramsparams, verifyFalse ) resp.raise_for_status() data resp.json() return { streams: data.get(data, []), count: len(data.get(data, [])) }4.4 第三步实现 Query 层工具mcp.tool() asyncdef query_logs( logql: str, start_hours: float 1.0, end_hours: float 0.0, limit: int 100, direction: str backward ) - dict: 按时间范围查询日志流。这是最常用的查询工具。 Args: logql: LogQL 查询语句例如 {apporder-service} | ERROR {namespaceproduction, levelerror} | json | duration 1s start_hours: 查询起点N 小时前默认 1 小时前 end_hours: 查询终点N 小时前默认 0即现在 limit: 最多返回条数默认 100最大 {MAX_LIMIT} direction: 排序方向backward最新的在前或 forward if direction notin (backward, forward): raise ValueError(direction 必须是 backward 或 forward) limit min(limit, MAX_LIMIT) start_ns, end_ns _validate_time_range(start_hours, end_hours) params { query: logql, start: start_ns, end: end_ns, limit: str(limit), direction: direction } asyncwith httpx.AsyncClient(timeout30.0) as client: resp await client.get( f{LOKI_URL}/loki/api/v1/query_range, headers_build_headers(), paramsparams, verifyFalse ) resp.raise_for_status() data resp.json() # 把 Loki 返回的格式整理成更易读的结构 result_streams data.get(data, {}).get(result, []) logs [] for stream in result_streams: labels stream.get(stream, {}) for ts_ns, log_line in stream.get(values, []): # 把 nanosecond 时间戳转成可读时间 ts_sec int(ts_ns) / 1e9 readable_ts datetime.fromtimestamp( ts_sec, tztimezone.utc ).strftime(%Y-%m-%d %H:%M:%S.%f UTC) logs.append({ timestamp: readable_ts, labels: labels, line: log_line }) return { query: logql, total: len(logs), logs: logs } mcp.tool() asyncdef instant_query(logql: str, time_hours_ago: float 0.0) - dict: 在某个特定时间点执行 LogQL 表达式通常用于 metric 类查询。 Args: logql: LogQL metric 表达式例如 count_over_time({apporder-service}[5m]) rate({apppayment-service, levelerror}[1m]) time_hours_ago: 在多少小时前的时间点执行0 表示当前时间 if time_hours_ago MAX_HOURS: raise ValueError(f时间点不能超过 {MAX_HOURS} 小时前) now datetime.now(timezone.utc) query_time now - timedelta(hourstime_hours_ago) query_time_ns str(int(query_time.timestamp() * 1e9)) asyncwith httpx.AsyncClient(timeout15.0) as client: resp await client.get( f{LOKI_URL}/loki/api/v1/query, headers_build_headers(), params{query: logql, time: query_time_ns}, verifyFalse ) resp.raise_for_status() data resp.json() return { query: logql, result_type: data.get(data, {}).get(resultType), result: data.get(data, {}).get(result, []) } # --- 入口 --- if __name__ __main__: # stdio 模式Claude Code 会把这个进程作为子进程启动 mcp.run(transportstdio)4.5 配置 Claude Code 加载这个 MCP编辑 Claude Code 的配置文件~/.claude.json或项目目录的.claude/settings.json{ mcpServers: { loki-logs: { command: python, args: [-m, loki_mcp.server], env: { LOKI_URL: http://your-loki-server:3100, LOKI_BEARER_TOKEN: your-token-here, LOKI_ORG_ID: your-org-id, LOKI_MAX_LIMIT: 500, LOKI_MAX_HOURS: 48 } } } }“凭证安全env字段里的 token 虽然是明文但这个配置文件只在本地机器上不提交到 git。比把 token 写死在代码里安全得多。如果公司有统一的 secret managerVault、AWS Secrets Manager 等可以改成调用 CLI 读取的方式。4.6 验证让 Claude Code 做一次日志排查配置好之后在 Claude Code 里可以这样用/mcp # 查看 MCP 工具列表确认 loki-logs 加载成功然后就可以直接提问最近一小时订单服务有哪些 ERROR 日志帮我分析一下根因。Claude Code 会自动调用list_labels→list_label_values(app)→query_logs这个链路不需要你手动拼 LogQL也不需要你去 Grafana 粘贴日志片段。demo 跑通了但生产环境还有几个坑要绕过。五、进阶案例三个让你重写 MCP 的生产教训教训一AI 不知道你的 LogQL 语法它会猜我们内部接好 Loki MCP 的第二天一个工程师让 Claude Code 查了一个服务的日志AI 生成的 LogQL 是{apppayment-service, errortrue}这个查询在 Loki 里返回空结果。问题是我们的日志里没有error这个 label错误信息在日志行的 JSON 内容里不是 label。正确写法应该是{apppayment-service} | json | level errorAI 不是不会 LogQL而是它不知道你的日志结构。解决方案在 MCP 工具的 docstring 里加一段当前系统的日志结构说明以及 2-3 个真实的查询示例。这是最低成本的做法。更系统化的做法是增加一个get_schema工具返回当前系统常用的 label 组合和日志格式说明供 AI 在生成查询前先读取。mcp.tool() asyncdef get_log_schema() - dict: 返回当前日志系统的结构说明和常用查询示例。 在构造复杂查询之前先调用这个工具了解日志格式。 return { label_structure: { app: 服务名如 order-service, payment-service, gateway, env: 环境值为 production 或 staging, namespace: K8s namespace, level: 注意level 不是 label需要通过 json 解析| json | level \error\ }, common_queries: [ {apporder-service} | json | level error, {apppayment-service, envproduction} | timeout | json, count_over_time({apporder-service} | json | level error [5m]) ], gotchas: [ 日志内容是 JSON 格式level/trace_id 等字段需要先 | json 再用, 时间戳在 Loki 里精度到纳秒API 返回的是纳秒字符串 ] }教训二不限制返回量AI 会把服务器打挂我们有条服务在出问题时每秒产生几百条日志。一次 AI 查了最近 24 小时的所有 ERROR 日志没有加任何过滤结果 Loki 开始扫描几十 GB 的数据查询超时、Loki 内存飙升影响了同时在看日志的其他工程师。这就是为什么要在代码里硬编码MAX_LIMIT和MAX_HOURS并且通过环境变量让管理员可以调整但AI 本身不能改变这两个限制——工具的参数不暴露这两个值。另一个建议是给query_logs加一个预检步骤先用count_over_time评估查询量超过阈值就提示 AI 缩小范围再查# 在 query_logs 执行前先做量级评估伪代码 count_query fcount_over_time(({logql})[{time_range}]) count_result await _loki_instant_query(count_query) if count_result WARN_THRESHOLD: return { warning: f该查询预计返回 {count_result} 条日志建议缩小时间范围或增加过滤条件, suggestion: 缩小时间范围到 1 小时内或增加 levelerror 过滤 }教训三stdio 在内网部署时的一个反直觉好处我们最开始按标准做法想把 MCP server 部署成一个 HTTP 服务然后让团队里所有工程师共享。结果遇到了需要管理 OAuth搭一套 Keycloak 比想象中麻烦MCP server 部署在内网某台机器上需要开防火墙端口每个工程师的 Claude Code 连同一个 MCP server有一个人把 Loki 打挂了所有人都受影响最后我们改回了 stdio 模式每个工程师本地起自己的 MCP 进程。收益是零运维负担没有额外服务要维护每个工程师的查询互相隔离一个人出问题不影响他人认证 token 管理各自负责不需要团队共享凭证stdio 的使用限制你的机器需要能直接访问 Loki/ES 的 HTTP 接口。如果有网络隔离只有特定跳板机能访问日志系统就需要走 HTTP 模式 部署在跳板机上。三条教训总结工具粒度 → 查询限制 → 部署模式这三件事想清楚了一个能在生产环境长期跑的日志 MCP 才算搭好了。六、小结今天这一讲我们做了一件事把企业日志系统包装成 MCP让 Claude Code 从等你粘贴日志升级到主动去查日志。如果你只能记住三件事记住这三件工具分层比工具数量更重要——Discovery 层探索结构和 Query 层精确查询分开让 AI 的查询行为可预期、可控制。内网 stdio 优于 HTTP——少于 10 人的团队用 stdio 模式零运维、各自隔离、认证简单只有需要共享 MCP server 时才上 HTTP OAuth。MCP 层是安全的最后防线——查询上限、时间范围限制、label 白名单都在 MCP 代码里硬编码AI 看到的工具参数里不暴露这些限制的调整权限。这一讲的 Loki MCP 代码在第 32 讲会作为 RCA 自动化 Skill 的核心组件再次登场——那一讲我们把查日志 分析根因 更新 runbook 这整个流程串成一个 Skill你会看到今天打的这个地基有多重要。