Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/search_tools.py:263-306 (handler)

  Main handler function that retrieves full project details by exact project name. Internally uses td_find_project for lookup and then fetches complete project object.
  async def td_get_project_by_name( project_name: str, ) -> dict[str, Any]: """Get full project details using exact name instead of ID. Convenient shortcut when you know the exact project name. Combines find + get operations for immediate detailed results. Common scenarios: - User provides exact project name, need full details - Quick project metadata lookup by name - Avoiding two-step process (find ID then get details) - Getting revision/timestamps for known project Requires exact name match. For fuzzy search use td_find_project. Returns same details as td_get_project but using name lookup. """ if not project_name or not project_name.strip(): return _format_error_response("Project name cannot be empty") # Use find_project with exact match search_result = await td_find_project(project_name, exact_match=True) if search_result.get("found") and search_result.get("projects"): project = search_result["projects"][0] # Get full details using td_get_project client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: full_project = client.get_project(project["id"]) if full_project: return {"project": full_project.model_dump()} else: return _format_error_response( f"Could not retrieve details for project '{project_name}'" ) except Exception as e: return _format_error_response(f"Failed to get project details: {str(e)}") return _format_error_response(f"Project '{project_name}' not found")
• td_mcp_server/search_tools.py:16-32 (registration)

  Registration function that decorates td_get_project_by_name with mcp.tool() to register it as an 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)
• td_mcp_server/search_tools.py:33-153 (helper)

  Helper function td_find_project used internally by td_get_project_by_name for exact name matching.
  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)}")