pip install mcp

#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_versinotallow="0.4.1",
                    capabilities=server.get_capabilities(
                        notification_optinotallow=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )

    asyncio.run(_run())

if __name__ == "__main__":
    # 直接調(diào)用main函數(shù)
    main()

#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",
                descriptinotallow="通過ID或URL獲取Sentry問題詳情",
                arguments=[
                ],
            ),
            types.Prompt(
                name="session-stats",
                descriptinotallow="獲取Sentry會話統(tǒng)計數(shù)據(jù)和崩潰率",
                arguments=[
                  
                ],
            ),
             types.Prompt(
                name="apk-history",
                descriptinotallow="獲取APK檢測歷史數(shù)據(jù)",
                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

#server.py
asyncdef 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()

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

        # 獲取最新事件并生成堆棧跟蹤
        latest_event = hashes[0]["latestEvent"]
        stacktrace = create_stacktrace(latest_event)

        # 返回格式化的問題數(shù)據(jù)
        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)}")

#server.py
asyncdef 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:
        # 構(gòu)建查詢參數(shù)
        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 isnotNone:
            params["includeTotals"] = include_totals
        if include_series isnotNone:
            params["includeSeries"] = include_series
        if query:
            params["query"] = query
            
        # 發(fā)送請求
        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(
            organizatinotallow=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)}")

#server.py
asyncdef 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響應(yīng)結(jié)構(gòu)中獲取數(shù)據(jù)
    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)}")

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

Sentry問題分析報告
我已經(jīng)查詢到關(guān)于您提供的Sentry問題的詳細信息。這是一個發(fā)生在flavorsOnline_arm64環(huán)境中的項目6的錯誤：
錯誤概覽
錯誤類型：RemoteServiceException$CannotDeliverBroadcastException
錯誤描述：can't deliver broadcast
問題ID：644
狀態(tài)：未解決
嚴(yán)重級別：致命(fatal)
首次出現(xiàn)：2024-10-03 04:36:33
最近出現(xiàn)：2025-03-21 10:44:02
事件計數(shù)：18次
堆棧跟蹤
錯誤發(fā)生在Android系統(tǒng)的廣播傳遞機制中，堆棧跟蹤顯示：
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
崩潰率分析
根據(jù)過去7天的會話統(tǒng)計：
崩潰率：約0.0001 (0.01%)
無崩潰率：約0.9999 (99.99%)
總會話數(shù)：1,239,555
雖然崩潰率較低，但由于這是一個致命錯誤，且在近期(3月21日)仍有發(fā)生，應(yīng)該引起關(guān)注。
問題分析
錯誤性質(zhì)：這是一個與Android廣播機制相關(guān)的RemoteServiceException，表明應(yīng)用嘗試發(fā)送一個廣播，但系統(tǒng)無法傳遞。
可能原因：
廣播接收器不存在或已被卸載
目標(biāo)組件權(quán)限問題
系統(tǒng)資源限制導(dǎo)致廣播傳遞失敗
特定于arm64架構(gòu)的問題
影響范圍：
錯誤發(fā)生在flavorsOnline_arm64環(huán)境
相對較低的崩潰率(0.01%)表明這可能是特定設(shè)備或特定條件下才會觸發(fā)的問題
建議解決方案
防御性編程：
在發(fā)送廣播前檢查接收器是否存在
使用try-catch塊包裝廣播發(fā)送代碼
錯誤位置定位：
檢查ActivityThread.java:2087處的throwRemoteServiceException方法調(diào)用點
找出應(yīng)用中哪些地方發(fā)送了可能觸發(fā)此異常的廣播
設(shè)備/版本針對性修復(fù)：
由于這個問題僅在arm64架構(gòu)出現(xiàn)，可以考慮為該架構(gòu)添加特定的兼容性代碼
監(jiān)控計劃：
雖然崩潰率低，但應(yīng)繼續(xù)監(jiān)控，特別是在新版本發(fā)布后
由于這個錯誤是與Android系統(tǒng)廣播機制相關(guān)的，建議查看應(yīng)用中所有發(fā)送廣播的代碼，尤其是在6.6.0版本中新增或修改的部分，確保它們都有適當(dāng)?shù)腻e誤處理機制。
需要我進一步分析這個問題的特定方面嗎？