用 MCP 扩展 Cursor 能力：解锁 Sentry 服务数据新玩法

一、MCP 简介

MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 推出的一种开放标准，旨在统一大型语言模型（LLM）与外部数据源和工具之间的通信协议。

使用模型上下文协议（MCP）可以构建服务器，向LLM应用程序公开数据和功能。MCP 服务器可以完成的功能如下：

  • 通过资源（Resources）暴露数据（类似 GET 请求的执行接口；它们用于将信息加载到 LLM 的上下文中）
  • 通过工具（Tools）提供功能（类似 POST 请求的执行接口；它们用于执行代码或者产生其他附带影响）
  • 通过提示（Prompts）定义交互模式（类似一个服务列表；可以用来查询服务器提供哪些服务，需要哪些参数等）

MCP 的主要目的在于解决当前 AI模型因数据孤岛限制而无法充分发挥潜力的难题。通过 MCP，AI应用能够安全地访问和操作本地及远程数据，为 AI 应用提供了连接万物的接口。

在当前 AI应用开发中，我们经常需要让 AI模型与现有的业务系统进行交互。例如，查询数据库、调用监控系统 API 、分析日志等。MCP 提供了一种标准化的方式来构建这种交互，使 AI 模型能够"看见"外部世界的数据。

二、MCP 的核心优势

  1. 能力扩展：使 AI 模型能够超越其训练数据的限制，获取实时外部信息
  2. 标准化接口：提供统一的接口规范，简化服务集成过程
  3. 安全可控：通过明确的权限管理和访问控制，保障数据安全
  4. 低代码集成：减少编写胶水代码的需求，降低开发成本
  5. 实时数据访问：让 AI能够基于最新数据做出决策和回复

三、MCP可以实现的功能举例

以下服务是官方提供或者第三方通过MCP协议实现的一些服务工具。由TypeScript或者 Python SDK实现，详细说明以及更多例子可以参见官方文档：官方文档

  • AWS 知识库检索 - 使用Bedrock Agent Runtime 从 AWS 知识库中检索信息
  • Brave 搜索 - 使用 Brave 的搜索 API进行网络和本地搜索
  • EverArt - 使用各种模型进行 AI图像生成
  • Everything - 参考/测试服务器，包含提示、资源和工具
  • Fetch - 网络内容获取和转换，提高 LLM 使用效率
  • 文件系统 - 具有可配置访问控制的安全文件操作
  • Git- 用于读取、搜索和操作 Git仓库的工具
  • GitHub - 仓库管理、文件操作和 GitHub API集成
  • GitLab - GitLab API，支持项目管理
  • Google Drive - Google Drive 文件访问和搜索功能
  • Google 地图 - 位置服务、路线指引和地点详情
  • 记忆系统 - 基于知识图谱的持久化记忆系统
  • PostgreSQL - 具有架构检查功能的只读数据库访问
  • Puppeteer - 浏览器自动化和网页抓取
  • Redis - 与 Redis 键值存储交互
  • Sentry - 从 Sentry.io 检索和分析问题
  • Sequential Thinking - 通过思维序列进行动态和反思性问题解决
  • Slack- 频道管理和消息发送功能
  • SQLite - 数据库交互和商业智能功能

三、以 Sentry 服务为例的 MCP 实践

3.1 背景与实现的功能

在我们的项目中，我们使用Sentry 作为 bug上报平台。数据可以通过接口获取，其与 MCP服务结合，能够便捷地将数据接入 AI，可以更方便的分析 crash原因和查询崩溃率等数据。官方提供的 demo 程序，更是为我们进一步扩展 Sentry工具功能提供了很好的参考示例。
借助 Sentry接入 MCP协议，我们可以只将 bug ID或问题链接提交给 cursor，便能迅速获取详细的 bug 信息以及可能的解决方案，这可以显著提升问题排查与解决的效率。

同时，我们把 Sentry统计信息也接入系统，赋予了 cursor访问各版本崩溃率数据以及全版本综合崩溃率数据的权限。AI能够据此输出直观的对比图表或者数据。

不仅如此，为进一步丰富 Sentry MCP 工具的功能，我们还将后端应用性能监控（APM）数据整合至该服务中。如此一来，cursor可直接查询到更为全面、丰富的应用信息，全方位提升了数据的可获取性与利用价值。

3.2 Sentry MCP服务组件功能结构

Sentry MCP服务组件功能结构图如下：

图1
  • 程序流程
    程序入口：通过main函数启动程序，接收Sentry和SNS身份验证令牌
    服务器创建：serve函数创建MCP服务器并注册各种处理程序
    请求处理：当 AI助手发起工具调用请求时，handle_call_tool函数依据工具名称进行请求的路由分发
    数据获取：专门的处理函数通过外部API获取数据，并将其封装在相应的数据类中
    响应生成：数据类提供方法将数据转换为所需的格式，返回给AI助手
  • 主要组件
    数据类：
    SentryIssueData：存储Sentry错误和crash信息，包括标题、ID、状态和堆栈跟踪
    SentrySessionStatsData：存储Sentry会话统计数据，如崩溃率、用户数等
    ApkHistoryData：存储APK检测历史数据，包括版本、大小和各组件信息
    处理函数：
    handle_sentry_issue：获取特定Sentry问题详情
    handle_session_stats：获取会话统计数据，如崩溃率和稳定性指标
    handle_apk_history：获取APK历史数据，包括大小和组件分析
    服务器配置：
    serve：配置MCP服务器，注册工具、提示和处理程序
    handle_call_tool：分发工具调用请求到相应的处理函数

3.3 功能实现

MCP协议有Python、Java、Kotlin等语言的SDK，由于Python使用最为普遍，我们这里使用PythonSDK来实现。

1. 安装MCP

pip install mcp

2. 功能入口

创建server.py文件作为服务程序的代码文件，以下所列代码都是在这个代码文件中。下面main函数是程序入口函数

#server.py
def main(auth_token: str, sns_token: str = None):
    async def _run():
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            server = await serve(auth_token, sns_token)
            await server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="sentry",
                    server_version="0.4.1",
                    capabilities=server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )

    asyncio.run(_run())

if __name__ == "__main__":
    # 直接调用main函数
    main()

main函数的功能如下：

  • 创建基于stdio的服务器通信通道
  • serve函数初始化MCP服务器实例
  • 配置服务器初始化选项
  • 启动服务器运行循环

3. 服务器实例创建

下面是简化版的serve函数：

#server.py
async def serve(auth_token: str, sns_token: str = None) -> Server:
    server = Server("sentry")
    http_client = httpx.AsyncClient(base_url=SENTRY_API_BASE)
    apk_http_client = httpx.AsyncClient(base_url=APK_API_BASE)

    @server.list_prompts()
    async def handle_list_prompts() -> list[types.Prompt]:
        prompts = [
            types.Prompt(
                name="sentry-issue",
                description="通过ID或URL获取Sentry问题详情",
                arguments=[
                ],
            ),
            types.Prompt(
                name="session-stats",
                description="获取Sentry会话统计数据和崩溃率",
                arguments=[
                  
                ],
            ),
             types.Prompt(
                name="apk-history",
                description="获取APK检测历史数据",
                arguments=[]
            )
            
        ]
            
        return prompts

    @server.get_prompt()
    async def handle_get_prompt(
        name: str, arguments: dict[str, str] | None
    ) -> types.GetPromptResult:
        if name == "sentry-issue":
             ...
            return issue_data.to_prompt_result()

        elif name == "session-stats":
            ...
            return stats_data.to_prompt_result()

        elif name == "apk-history":
           ...
            return history_data.to_prompt_result()
        else:
            raise ValueError(f"Unknown prompt: {name}")

    @server.list_tools()
    async def handle_list_tools() -> list[types.Tool]:
        tools = [
            types.Tool(
                name="get_sentry_issue",
                 
            ),
            types.Tool(
                name="get_sentry_session_stats",
               
            ) ,
            types.Tool(
                name="get_apk_history",
             
            )
        ]
        
      
            
        return tools

    @server.call_tool()
    async def handle_call_tool(
        name: str, arguments: dict | None
    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
        if name == "get_sentry_issue":
          
            return issue_data.to_tool_result()
        
        elif name == "get_sentry_session_stats":
        
            return stats_data.to_tool_result()
            
        elif name == "get_apk_history":
         
            return history_data.to_tool_result()
        
        else:
            raise ValueError(f"Unknown tool: {name}")

    return server

serve函数是这个MCP Sentry服务的核心组件，负责创建、配置和返回一个完整的MCP服务器实例。它通过四个主要组件构建了完整服务：

  • 提示列表：定义可用的查询服务（问题详情、崩溃率统计、APK历史）
  • 提示处理器：执行查询逻辑并返回结果
  • 工具列表：注册可调用的工具集合
  • 工具调用处理器：处理具体工具调用并获取数据
    整体功能是将Sentry和APK监控系统的复杂API调用封装成结构化接口，方便AI助手获取崩溃报告、稳定性数据和APK分析信息。

3.4 三个查询功能的实现

1. Sentry问题详情获取功能实现

handle_sentry_issue函数是MCP Sentry服务的核心组件之一，它通过问题ID或URL获取Sentry问题的详细信息，包括问题标题、状态、级别、首次和最后出现时间、事件数量，以及最关键的堆栈跟踪信息。下面的问题详情的获取主函数功能：

#server.py
async def handle_sentry_issue(
    http_client: httpx.AsyncClient, auth_token: str, issue_id_or_url: str
) -> SentryIssueData:
    try:
        # 提取问题ID
        issue_id = extract_issue_id(issue_id_or_url)

        # 获取问题详情
        response = await http_client.get(
            f"organizations/xxx/issues/{issue_id}/", 
            headers={"Authorization": f"Bearer {auth_token}"}
        )
        if response.status_code == 401:
            raise McpError(
                "Error: Unauthorized. Please check your MCP_SENTRY_AUTH_TOKEN token."
            )
        response.raise_for_status()
        issue_data = response.json()

        # 获取问题哈希值
        hashes_response = await http_client.get(
            f"organizations/xxx/issues/{issue_id}/hashes/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )
        hashes_response.raise_for_status()
        hashes = hashes_response.json()

        if not hashes:
            raise McpError("No Sentry events found for this issue")

        # 获取最新事件并生成堆栈跟踪
        latest_event = hashes[0]["latestEvent"]
        stacktrace = create_stacktrace(latest_event)

        # 返回格式化的问题数据
        return SentryIssueData(
            title=issue_data["title"],
            issue_id=issue_id,
            status=issue_data["status"],
            level=issue_data["level"],
            first_seen=issue_data["firstSeen"],
            last_seen=issue_data["lastSeen"],
            count=issue_data["count"],
            stacktrace=stacktrace
        )

    except SentryError as e:
        raise McpError(str(e))
    except httpx.HTTPStatusError as e:
        raise McpError(f"Error fetching Sentry issue: {str(e)}")
    except Exception as e:
        raise McpError(f"An error occurred: {str(e)}")

为了简化使用并确保数据一致性，我们在最新版本中对环境和项目做了固定配置：

  • 环境（environment）：固定为 ["flavorsOnline_arm64"]
  • 项目 ID（project）：固定为 [6]

这样用户在查询时不需要每次都指定这些参数。

2. Sentry会话统计数据获取功能实现

handle_session_stats函数是Sentry MCP服务的另一个核心组件，专门用于获取应用的会话统计数据，包括崩溃率、用户数量、会话持续时间等关键指标。这些数据对于监控应用稳定性、分析版本质量和追踪用户体验至关重要。

#server.py
async def handle_session_stats(
    http_client: httpx.AsyncClient, 
    auth_token: str, 
    organization: str,
    field: list[str],
    start: str = None,
    end: str = None,
    environment: list[str] = None,
    stats_period: str = None,
    project: list[int] = None,
    per_page: int = None,
    interval: str = None,
    group_by: list[str] = None,
    order_by: str = None,
    include_totals: int = 1,
    include_series: int = 1,
    query: str = None
) -> SentrySessionStatsData:
    try:
        # 构建查询参数
        params = {"field": field}
        
        if start:
            params["start"] = start
        if end:
            params["end"] = end
        if environment:
            params["environment"] = environment
        if stats_period:
            params["statsPeriod"] = stats_period
        if project:
            params["project"] = project
        if per_page:
            params["per_page"] = per_page
        if interval:
            params["interval"] = interval
        if group_by:
            params["groupBy"] = group_by
        if order_by:
            params["orderBy"] = order_by
        if include_totals is not None:
            params["includeTotals"] = include_totals
        if include_series is not None:
            params["includeSeries"] = include_series
        if query:
            params["query"] = query
            
        # 发送请求
        response = await http_client.get(
            f"organizations/{organization}/sessions/",
            headers={"Authorization": f"Bearer {auth_token}"},
            params=params
        )
        
        if response.status_code == 401:
            raise McpError(
                "Error: Unauthorized. Please check your MCP_SENTRY_AUTH_TOKEN token."
            )
        
        response.raise_for_status()
        data = response.json()
        
        stats_data = SentrySessionStatsData(
            organization=organization,
            start_time=data.get("start", ""),
            end_time=data.get("end", ""),
            intervals=data.get("intervals", []),
            groups=data.get("groups", []),
            query=data.get("query", "")
        )
        report_file_path = generate_crash_rate_html_report(
                    data=stats_data,
                    params=params
                )
                stats_data.report_file_path = report_file_path

        return stats_data
    except httpx.HTTPStatusError as e:
        raise McpError(f"Error fetching Sentry session statistics: {str(e)}")
    except Exception as e:
        raise McpError(f"An error occurred: {str(e)}")
  • 功能亮点
    灵活的查询参数：支持多种筛选条件，如时间范围、环境、项目ID、查询表达式等
    丰富的统计指标：提供多种会话相关指标，如：
    会话总数 ( sum(session))
    独立用户数 (count_unique(user))
    平均会话时长 (avg(session.duration))
    会话时长分位数 (p50/p75/p90/p95/p99/max(session.duration))
    用户崩溃率 (crash_rate(user))
    会话崩溃率 (crash_rate(session))
    用户崩溃免除率(crash_free_rate(user))
    会话崩溃免除率 (crash_free_rate(session))
    数据系列和汇总：同时支持时间序列数据和汇总数据，生成html文件，便于趋势分析和整体评估
    分组和排序：支持对数据进行分组和排序，便于比较分析不同维度的数据

通过这一功能，使用cursor能够帮助我们监控和分析应用的稳定性表现，及时发现潜在问题，提供数据支持以优化用户体验和应用质量。

3. APK检测历史数据获取功能实现

handle_apk_history函数是Sentry MCP服务中专门用于获取和分析APK检测历史数据的组件。它能够追踪应用各版本的体积变化、组件大小分布和版本迭代情况。

#server.py
async def handle_apk_history(
    http_client: httpx.AsyncClient, 
    sns_token: str
) -> ApkHistoryData:
    try:
        response = await http_client.get(
            "getHistory/0",
            headers={"sns_token": sns_token}
        )
        
        if response.status_code == 401:
            raise McpError(
                "Error: Unauthorized. Please check your SNS token."
            )
        
        response.raise_for_status()
        data = response.json()
        
        # 从新的API响应结构中获取数据
    apkData = ApkHistoryData(
            apk_list=data.get("data", {}).get("apkList", []),
            total_count=data.get("data", {}).get("totalCount", 0)
        )
     report_file_path = generate_apk_history_html_report(
                    data=apkData
                )
                apkData.report_file_path = report_file_path


         
        return apkData
    except httpx.HTTPStatusError as e:
        raise McpError(f"Error fetching APK history data: {str(e)}")
    except Exception as e:
        raise McpError(f"An error occurred: {str(e)}")

这些信息源自我们内部 应用性能监控（APM）接口，它详细呈现了 APK的相关内容，具体涵盖：

  • 基本信息：APK 的版本、版本号以及名称。
  • 大小详情：APK的总大小，以及实际下载大小。
  • 组件分析：对 APK 各组件，即Assets、Dex、Lib以及资源部分的大小进行剖析。
  • 元数据：包含时间戳、文件路径、备注信息，以及特性描述等。
    此外，这些数据以结构化形式整合，输出到html文件中，将复杂信息有序组织，方便后续进行深入分析与直观展示 。
4. 在cursor中集成

在cursor的设置中添加MCP服务，Type选择command，然后在Command中输入python执行程序，保存后，可以在cursor对话框中使用，添加的界面如图所示：

图2

选择Type时，还有另一种Type是sse，sse是服务器发送事件（Server-Sent Events，SSE）它是一种允许服务器向客户端实时推送数据的 Web API。SSE建立了一个单向的HTTP连接，服务器可以在有新数据时随时向客户端发送消息，而客户端只需监听服务器的推送。我们Sentry MCP这里只使用了主动获取数据的方式，因此这里选择了command命令行的方式。
command完整的命令如下所示：

python3 /Users/xxx/server.py --auth-token xxx sns-token xxx

其中xxx是我们sentry平台和应用性能监控（APM）的接口token

3.4 效果展示

1. 查询崩溃率数据

使用handle_session_stats函数可以轻松获取应用的崩溃率数据，例如在cursor中输入：帮我查询6.6.0版本和6.5.0版本最近14天的崩溃率详情，输出html结果如下：

图3
图4
图5
图6

2.查询bug详情

在cursor中输入：帮我查询这个bug的详细信息644，请求cursor结果如下：

Sentry问题分析报告
我已经查询到关于您提供的Sentry问题的详细信息。这是一个发生在flavorsOnline_arm64环境中的项目6的错误：
错误概览
错误类型：RemoteServiceException$CannotDeliverBroadcastException
错误描述：can't deliver broadcast
问题ID：644
状态：未解决
严重级别：致命(fatal)
首次出现：2024-10-03 04:36:33
最近出现：2025-03-21 10:44:02
事件计数：18次
堆栈跟踪
错误发生在Android系统的广播传递机制中，堆栈跟踪显示：
ZygoteInit.java:954 in main
RuntimeInit.java:559 in run
Method.java:None in invoke
ActivityThread.java:8395 in main
Looper.java:300 in loop
Looper.java:211 in loopOnce
Handler.java:106 in dispatchMessage
ActivityThread.java:2349 in handleMessage
None:0 in -$$Nest$mthrowRemoteServiceException
ActivityThread.java:2087 in throwRemoteServiceException
崩溃率分析
根据过去7天的会话统计：
崩溃率：约0.0001 (0.01%)
无崩溃率：约0.9999 (99.99%)
总会话数：1,239,555
虽然崩溃率较低，但由于这是一个致命错误，且在近期(3月21日)仍有发生，应该引起关注。
问题分析
错误性质：这是一个与Android广播机制相关的RemoteServiceException，表明应用尝试发送一个广播，但系统无法传递。
可能原因：
广播接收器不存在或已被卸载
目标组件权限问题
系统资源限制导致广播传递失败
特定于arm64架构的问题
影响范围：
错误发生在flavorsOnline_arm64环境
相对较低的崩溃率(0.01%)表明这可能是特定设备或特定条件下才会触发的问题
建议解决方案
防御性编程：
在发送广播前检查接收器是否存在
使用try-catch块包装广播发送代码
错误位置定位：
检查ActivityThread.java:2087处的throwRemoteServiceException方法调用点
找出应用中哪些地方发送了可能触发此异常的广播
设备/版本针对性修复：
由于这个问题仅在arm64架构出现，可以考虑为该架构添加特定的兼容性代码
监控计划：
虽然崩溃率低，但应继续监控，特别是在新版本发布后
由于这个错误是与Android系统广播机制相关的，建议查看应用中所有发送广播的代码，尤其是在6.6.0版本中新增或修改的部分，确保它们都有适当的错误处理机制。
需要我进一步分析这个问题的特定方面吗？

3.查询应用性能监控（APM）中数据

在cursor中输入：帮我查询6.6.0的性能数据，并和6.5.0版本对比。请求cursor后生成的html结果如下：

图7
图8
图9

通过结果，我们可以看出，通过 MCP协议，本地数据得以充分供给给 AI，成功赋予了AI访问本地数据的能力。这使得 AI能够深度挖掘并高效利用这些数据。

四、总结

本文对 MCP 进行了简要介绍。MCP是一种高效且通用的协议，它能让AI安全、高效地访问本地数据。我们以 Sentry接入MCP为例，成功实现了Sentry监控服务与cursor助手的集成，达成了应用崩溃率的智能分析以及crash问题的详细查询，助力大家更好地理解 MCP 的工作模式。

希望本文能帮助读者更好地认识 MCP 协议，熟练掌握将本地数据接入AI的方法。期待这篇技术分享能对您了解和运用 MCP有所助益。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容