get_tapd_stories
Fetch formatted JSON data for requirements from TAPD projects in a paginated manner, enabling efficient access to project management details for AI-driven analysis.
Instructions
获取TAPD平台指定项目的需求数据(支持分页)
Returns:
str: 格式化后的需求数据JSON字符串
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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