Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/url_tools.py:31-90 (handler)

  The core handler function for the 'td_analyze_url' tool. It parses input URL using regex to detect resource types (workflow, project, job), extracts IDs, and fetches details via API calls or delegates to td_get_workflow. Includes input validation and error handling.
  async def td_analyze_url(url: str) -> dict[str, Any]: """Analyze any Treasure Data console URL to get resource details. Smart URL parser that extracts IDs and fetches information. Use when someone shares a console link in Slack, email, or documentation. Common scenarios: - Someone shares workflow URL during incident investigation - Documentation contains console links to resources - Error message includes console URL reference - Quick lookup from browser URL copy/paste Supported formats: - Workflow: https://console.../app/workflows/12345678/info - Project: https://console.../app/projects/123456 - Job: https://console.../app/jobs/123456 Automatically detects type and returns full resource information. """ if not url or not url.strip(): return _format_error_response("URL cannot be empty") # Parse workflow URL workflow_match = re.search(r"/app/workflows/(\d+)", url) if workflow_match: workflow_id = workflow_match.group(1) return await td_get_workflow(workflow_id) # Parse project URL project_match = re.search(r"/app/projects/(\d+)", url) if project_match: project_id = project_match.group(1) client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: project = client.get_project(project_id) if project: return {"type": "project", "project": project.model_dump()} else: return _format_error_response( f"Project with ID '{project_id}' not found" ) except Exception as e: return _format_error_response(f"Failed to get project: {str(e)}") # Parse job URL job_match = re.search(r"/app/jobs/(\d+)", url) if job_match: job_id = job_match.group(1) return { "type": "job", "job_id": job_id, "message": "Job information retrieval not yet implemented", } return _format_error_response( "Unrecognized URL format. Supported: /app/workflows/ID, /app/projects/ID" )
• td_mcp_server/mcp_impl.py:710-710 (registration)

  Top-level registration call in the MCP server implementation that invokes url_tools.register_url_tools to add the td_analyze_url tool to the MCP instance.
  url_tools.register_url_tools(mcp, _create_client, _format_error_response)
• td_mcp_server/url_tools.py:19-29 (registration)

  The registration function that sets global dependencies and applies the mcp.tool() decorator to td_analyze_url, making it available as an MCP tool.
  def register_url_tools(mcp_instance, create_client_func, format_error_func): """Register URL tools with the provided MCP instance.""" global mcp, _create_client, _format_error_response mcp = mcp_instance _create_client = create_client_func _format_error_response = format_error_func # Register all tools mcp.tool()(td_analyze_url) mcp.tool()(td_get_workflow)
• td_mcp_server/url_tools.py:31-49 (schema)

  Input schema defined by function signature (url: str) and comprehensive docstring specifying supported URL formats and expected output structure (resource type and details). Output typed as dict[str, Any].
  async def td_analyze_url(url: str) -> dict[str, Any]: """Analyze any Treasure Data console URL to get resource details. Smart URL parser that extracts IDs and fetches information. Use when someone shares a console link in Slack, email, or documentation. Common scenarios: - Someone shares workflow URL during incident investigation - Documentation contains console links to resources - Error message includes console URL reference - Quick lookup from browser URL copy/paste Supported formats: - Workflow: https://console.../app/workflows/12345678/info - Project: https://console.../app/projects/123456 - Job: https://console.../app/jobs/123456 Automatically detects type and returns full resource information. """