Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:295-348 (handler)

  The core handler function for the 'td_list_projects' MCP tool. It creates a TreasureDataClient instance with workflow support, fetches projects via client.get_projects() with pagination parameters, filters out system projects by default (those with 'sys' metadata), and returns either project names/IDs (default) or full details (verbose=True). Handles API errors gracefully.
  @mcp.tool() async def td_list_projects( verbose: bool = False, limit: int = DEFAULT_LIMIT, offset: int = 0, all_results: bool = False, include_system: bool = False, ) -> dict[str, Any]: """List workflow projects to find data pipelines and scheduled jobs. Shows all workflow projects containing Digdag workflows, SQL queries, and Python scripts. Returns names/IDs for navigation or verbose=True for details. Common scenarios: - Discover available data processing workflows - Find specific project by browsing names - Get project IDs for detailed exploration - Audit workflow projects in the account - List user projects (exclude system with include_system=False) Projects contain .dig files defining scheduled data pipelines. """ client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: projects = client.get_projects( limit=limit, offset=offset, all_results=all_results ) # Filter out system projects (those with "sys" metadata) if not include_system: projects = [ p for p in projects if not any(meta.key == "sys" for meta in p.metadata) ] if verbose: # Return full project details return {"projects": [project.model_dump() for project in projects]} else: # Return only project names and ids return { "projects": [ {"id": project.id, "name": project.name} for project in projects ] } except (ValueError, requests.RequestException) as e: return _format_error_response(f"Failed to retrieve projects: {str(e)}") except Exception as e: return _format_error_response( f"Unexpected error while retrieving projects: {str(e)}" )
• td_mcp_server/api.py:156-175 (schema)

  Pydantic BaseModel defining the structure of a Treasure Data workflow Project, used for input/output validation and serialization in the td_list_projects tool's response (via project.model_dump()). Includes fields like id, name, revision, timestamps, archive info, and metadata.
  class Project(BaseModel): """ Model representing a Treasure Data workflow project. In Treasure Data, a workflow project is a container for workflow definitions, which typically include SQL queries and Digdag files (.dig) that define the workflow execution steps and dependencies. These workflows are used for data processing, analytics pipelines, and scheduled jobs. """ id: str name: str revision: str created_at: str = Field(..., alias="createdAt") updated_at: str = Field(..., alias="updatedAt") deleted_at: str | None = Field(None, alias="deletedAt") archive_type: str = Field(..., alias="archiveType") archive_md5: str = Field(..., alias="archiveMd5") metadata: list[Metadata] = []
• td_mcp_server/api.py:332-378 (helper)

  Supporting method in TreasureDataClient that performs the actual API request to fetch workflow projects from Treasure Data's workflow API, handles pagination (mapping limit/offset to 'count' param, client-side slicing), parses JSON response into Project Pydantic models, and returns the list used by the tool handler.
  def get_projects( self, limit: int = 30, offset: int = 0, all_results: bool = False, ) -> list[Project]: """ Retrieve a list of workflow projects with pagination support. Workflow projects in Treasure Data contain workflow definitions used for data processing and analytics. Each project typically includes SQL queries and Digdag (.dig) files that define workflow execution steps and dependencies. These workflows are executed on the Treasure Data platform for scheduled data pipelines, ETL processes, and other automation tasks. Note: The API uses 'count' parameter for limiting results, but this method provides limit/offset interface for consistency with other methods. Args: limit: Maximum number of projects to retrieve (defaults to 30) offset: Index to start retrieving from (defaults to 0) all_results: If True, retrieves all projects ignoring limit and offset Returns: A list of Project objects representing workflow projects Raises: requests.HTTPError: If the API returns an error response """ # The projects API uses 'count' parameter, not limit/offset # Request more data if offset is specified # Increased to 200 to cover all projects (currently ~135) count = 200 if all_results else min(offset + limit, 200) params = {"count": count} response = self._make_request( "GET", "projects", base_url=self.workflow_base_url, params=params ) all_projects = [Project(**project) for project in response.get("projects", [])] if all_results: return all_projects else: # Apply offset and limit on the client side end_index = min(offset + limit, len(all_projects)) return all_projects[offset:end_index]