Treasure Data MCP Server

Instructions

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.

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]