Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/search_tools.py:33-152 (handler)

  The main handler function for the 'td_find_project' tool. It searches for Treasure Data projects by name using client APIs, supporting both exact and fuzzy matching across direct projects and workflows.
  async def td_find_project( search_term: str, exact_match: bool = False, ) -> dict[str, Any]: """Find project by name when you don't know the exact ID. Searches all projects and returns matches. Useful when you know project name but need the ID for other operations like downloading archives. Common scenarios: - User mentions project name, need to find ID - Looking for projects containing specific keywords - Getting project ID before using td_download_project_archive - Finding multiple projects with similar names Use exact_match=True for precise name matching, False for fuzzy search. Returns project IDs, names, and metadata for all 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: # First, try to get projects directly (up to 200) projects = client.get_projects(limit=200, all_results=True) found_projects = [] search_lower = search_term.lower() for project in projects: project_name = project.name.lower() if exact_match: if project_name == search_lower: found_projects.append(project) else: if search_lower in project_name: found_projects.append(project) if found_projects: return { "found": True, "count": len(found_projects), "projects": [ { "id": p.id, "name": p.name, "created_at": p.created_at, "updated_at": p.updated_at, } for p in found_projects ], } # If not found in projects, search through workflows workflows = client.get_workflows(count=1000, all_results=True) project_map = {} for workflow in workflows: project_name = workflow.project.name project_id = workflow.project.id if exact_match: if project_name.lower() == search_lower: if project_id not in project_map: project_map[project_id] = { "id": project_id, "name": project_name, "workflow_count": 0, } project_map[project_id]["workflow_count"] += 1 else: if search_lower in project_name.lower(): if project_id not in project_map: project_map[project_id] = { "id": project_id, "name": project_name, "workflow_count": 0, } project_map[project_id]["workflow_count"] += 1 if project_map: # Get full project details for found projects projects_with_details = [] for project_id, project_info in project_map.items(): try: project = client.get_project(project_id) if project: projects_with_details.append( { "id": project.id, "name": project.name, "created_at": project.created_at, "updated_at": project.updated_at, "workflow_count": project_info["workflow_count"], } ) except Exception: # Fallback to basic info projects_with_details.append(project_info) return { "found": True, "count": len(projects_with_details), "projects": projects_with_details, "source": "workflows", } return { "found": False, "count": 0, "message": f"No projects found matching '{search_term}'", } except Exception as e: return _format_error_response(f"Failed to search projects: {str(e)}")
• td_mcp_server/search_tools.py:16-31 (registration)

  The registration function that decorates and registers td_find_project as an MCP tool using mcp.tool().
  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)