TAPD Data Fetcher

Instructions

Returns:
    str: 格式化后的缺陷数据JSON字符串

Implementation Reference

• tapd_mcp_server.py:176-198 (handler)

  MCP tool handler and registration for 'get_tapd_bugs'. Wraps the helper get_bug_msg to fetch TAPD bugs and serialize to JSON string.

  @mcp.tool()
  async def get_tapd_bugs(clean_empty_fields: bool = True) -> str:
      """获取TAPD平台指定项目的缺陷数据（支持分页）
      
      功能描述:
          - 从TAPD API获取指定项目的所有缺陷数据
          - 支持按状态、优先级等条件过滤
          - 自动处理API认证和错误
          - 数据不保存至本地，建议仅在数据量较小时使用
          
      返回数据格式:
          - 每个缺陷包含ID、标题、严重程度、状态、解决方案等字段
          - 数据已按JSON格式序列化，确保AI客户端可直接解析
          
      Returns:
          str: 格式化后的缺陷数据JSON字符串，包含中文内容
      """
      try:
          bugs = await get_bug_msg(clean_empty_fields=clean_empty_fields)
          return json.dumps(bugs, ensure_ascii=False, indent=2)
      except Exception as e:
          return f"获取缺陷数据失败：{str(e)}"

• tapd_data_fetcher.py:86-122 (helper)

  Supporting utility function that implements the core logic of fetching bugs from TAPD API via paginated requests using aiohttp, with authentication and optional cleaning of empty fields.

  async def get_bug_msg(clean_empty_fields: bool = True):
      url = 'https://api.tapd.cn/bugs'  # TAPD缺陷API地址
      bugs_list = []  # 存储所有缺陷数据的列表
      page = 1  # 初始页码
      while True:
          params = {
              'workspace_id': WORKSPACE_ID,
              # 若需要获取所有字段，请注释下面的fields参数
              # 'fields': 'id,title,description,priority,severity,module,status,reporter,created,bugtype,resolved,closed,modified,lastmodify,auditer,de,fixer,version_test,version_report,version_close,version_fix,baseline_find,baseline_join,baseline_close,baseline_test,sourcephase,te,current_owner,iteration_id,resolution,source,originphase,confirmer,milestone,participator,closer,platform,os,testtype,testphase,frequency,cc,regression_number,flows,feature,testmode,estimate,issue_id,created_from,release_id,verify_time,reject_time,reopen_time,audit_time,suspend_time,due,begin,deadline,in_progress_time,assigned_time,template_id,story_id,label,size,effort,effort_completed,exceed,remain,secret_root_id,priority_label,workspace_id',
              'page': page
          }
          async with aiohttp.ClientSession() as session:
              async with session.get(url, auth=aiohttp.BasicAuth(API_USER, API_PASSWORD), params=params) as response:
                  if response.status != 200:
                      print(f'获取缺陷第{page}页失败: {await response.text()}')
                      break
                  result = await response.json()
                  if result.get('status') != 1:
                      print(f'获取缺陷第{page}页失败: {result.get("info")}')
                      break
                  current_page_data = result.get('data', [])
                  if not current_page_data:  # 无更多数据时结束循环
                      break
                  for bug in current_page_data:  # 遍历当前页的每条缺陷数据，提取Bug字段
                      bug_data = bug.get('Bug', {})
                      if not bug_data.get('title'):  # 检查缺陷title是否为空
                          print(f'发现缺陷数据title为空（第{page}页），结束获取')
                          return bugs_list  # 遇到空值立即终止并返回已有数据
                      # 根据参数决定是否清洗空数据字段（None/空字符串）
                      if clean_empty_fields:
                          processed_bug = {k: v for k, v in bug_data.items() if v not in (None, "")}
                      else:
                          processed_bug = bug_data
                      bugs_list.append(processed_bug)
          page += 1  # 页码递增
      print(f'缺陷数据获取完成，共获取{len(bugs_list)}条')
      return bugs_list