Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/search_tools.py:154-260 (handler)

  The handler function that executes the td_find_workflow tool. It searches for workflows matching the search_term, optionally filtered by project_name, exact_match, and status_filter. Returns matching workflows with details including project info and latest execution status.
  async def td_find_workflow( search_term: str, project_name: str | None = None, exact_match: bool = False, status_filter: str | None = None, ) -> dict[str, Any]: """Find workflows by name to get IDs and check execution status. Essential for locating specific workflows when you know the name. Returns workflow IDs, project info, and latest execution status. Common scenarios: - User mentions workflow name, need to find details - Looking for failing workflows with specific names - Finding workflows within a specific project - Getting workflow ID before detailed analysis - Checking if a named workflow is running/failed Filters: project_name (optional), status ('success', 'error', 'running'). Use exact_match=True for precise names, False for partial matches. """ if not search_term or not search_term.strip(): return _format_error_response("Search term cannot be empty") client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: # Get workflows (up to 1000) workflows = client.get_workflows(count=1000, all_results=True) found_workflows = [] search_lower = search_term.lower() project_lower = project_name.lower() if project_name else None for workflow in workflows: workflow_name = workflow.name.lower() workflow_project = workflow.project.name.lower() # Check workflow name match name_match = False if exact_match: name_match = workflow_name == search_lower else: name_match = search_lower in workflow_name if not name_match: continue # Check project filter if specified if project_lower: if project_lower not in workflow_project: continue # Check status filter if specified if status_filter: if workflow.latest_sessions: last_status = workflow.latest_sessions[0].last_attempt.status if last_status != status_filter: continue else: # No sessions means no status, skip if filtering by status continue # Prepare workflow info workflow_info = { "id": workflow.id, "name": workflow.name, "project": { "id": workflow.project.id, "name": workflow.project.name, }, "timezone": workflow.timezone, "scheduled": workflow.schedule is not None, } # Add latest status if available if workflow.latest_sessions: latest = workflow.latest_sessions[0] workflow_info["latest_session"] = { "session_time": latest.session_time, "status": latest.last_attempt.status, "success": latest.last_attempt.success, } else: workflow_info["latest_session"] = None found_workflows.append(workflow_info) if found_workflows: return { "found": True, "count": len(found_workflows), "workflows": found_workflows, } message = f"No workflows found matching '{search_term}'" if project_name: message += f" in project '{project_name}'" if status_filter: message += f" with status '{status_filter}'" return {"found": False, "count": 0, "message": message} except Exception as e: return _format_error_response(f"Failed to search workflows: {str(e)}")
• td_mcp_server/search_tools.py:27-31 (registration)

  The registration of the td_find_workflow tool using the mcp.tool() decorator within the register_mcp_tools function.
  mcp.tool()(td_find_project) mcp.tool()(td_find_workflow) mcp.tool()(td_get_project_by_name) mcp.tool()(td_smart_search)