TAPD Data Fetcher

Instructions

Returns:
    str: 格式化后的需求数据JSON字符串

Implementation Reference

• tapd_mcp_server.py:153-175 (handler)

  MCP tool handler and registration for 'get_tapd_stories'. Fetches TAPD stories data via helper function and returns formatted JSON.

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

• tapd_data_fetcher.py:47-83 (helper)

  Helper function that implements the core logic of fetching TAPD stories via paginated API requests using aiohttp.

  async def get_story_msg(clean_empty_fields: bool = True):
      url = 'https://api.tapd.cn/stories'  # TAPD需求API地址
      stories_list = []  # 存储所有需求数据的列表
      page = 1  # 初始页码
      while True:
          params = {
              'workspace_id': WORKSPACE_ID,
              # 若需要获取所有字段，请注释下面的fields参数
              # 'fields': 'id,workitem_type_id,name,description,workspace_id,creator,created,modified,status,step,owner,cc,begin,due,size,priority,developer,iteration_id,test_focus,type,source,module,version,completed,category_id,path,parent_id,children_id,ancestor_id,level,business_value,effort,effort_completed,exceed,remain,release_id,bug_id,templated_id,created_from,feature,label,progress,is_archived,tech_risk,flows,priority_label',
              '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 story in current_page_data:  # 遍历当前页的每条需求数据，提取Story字段
                      story_data = story.get('Story', {})
                      if not story_data.get('id'):  # 检查需求id是否为空
                          print(f'发现需求数据id为空（第{page}页），结束获取')
                          return stories_list  # 遇到空值立即终止并返回已有数据
                      # 根据参数决定是否清洗空数据字段（None/空字符串）
                      if clean_empty_fields:
                          processed_story = {k: v for k, v in story_data.items() if v not in (None, "")}
                      else:
                          processed_story = story_data
                      stories_list.append(processed_story)
          page += 1  # 页码递增
      print(f'需求数据获取完成，共获取{len(stories_list)}条')
      return stories_list