Treasure Data MCP Server

Instructions

Implementation Reference

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

  Core implementation of the td_find_workflow tool. Searches all workflows for name matches (exact or partial), applies optional project_name and status_filter, and returns list of matching workflows with project info, schedule status, and latest execution details.
  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:16-31 (registration)

  Registers the td_find_workflow tool (and others) with the MCP instance using the mcp.tool() decorator in the register_mcp_tools function.
  def register_mcp_tools( mcp_instance, create_client_func, format_error_func, validate_project_func ): """Register MCP tools with the provided MCP instance.""" global mcp, _create_client, _format_error_response, _validate_project_id mcp = mcp_instance _create_client = create_client_func _format_error_response = format_error_func _validate_project_id = validate_project_func # Register all tools mcp.tool()(td_find_project) mcp.tool()(td_find_workflow) mcp.tool()(td_get_project_by_name) mcp.tool()(td_smart_search)
• td_mcp_server/search_tools.py:154-159 (schema)

  Function signature providing input schema (type hints for parameters) and output type (dict[str, Any]) for the tool, used by MCP framework for validation.
  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]: