td_list_projects
List workflow projects to discover data pipelines, scheduled jobs, and processing workflows. Shows project names and IDs for navigation or detailed exploration with verbose mode.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| all_results | No | ||
| include_system | No | ||
| limit | No | ||
| offset | No | ||
| verbose | No |
Implementation Reference
- td_mcp_server/mcp_impl.py:296-347 (handler)The core handler function for the 'td_list_projects' tool. Defined as an async function decorated with @mcp.tool(), which handles registration. Lists Treasure Data workflow projects with options for verbosity, pagination, and filtering system projects. Uses a client to fetch projects and formats the response accordingly.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)}" )