Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:150-190 (handler)

  The primary handler function td_list_databases decorated with @mcp.tool(). It creates a TreasureDataClient, calls get_databases on it, and returns either database names or full details based on verbose flag.
  @mcp.tool() async def td_list_databases( verbose: bool = False, limit: int = DEFAULT_LIMIT, offset: int = 0, all_results: bool = False, ) -> dict[str, Any]: """List available databases to find data sources and check access. Shows all databases you can access. Returns just names for quick overview, or set verbose=True for details like table count and permissions. Common scenarios: - Discover what databases are available in your account - Check permissions on specific databases - Get database list for documentation or auditing Use pagination (limit/offset) for large lists or all_results=True for everything. """ client = _create_client() if isinstance(client, dict): return client try: databases = client.get_databases( limit=limit, offset=offset, all_results=all_results ) if verbose: # Return full database details return {"databases": [db.model_dump() for db in databases]} else: # Return only database names return {"databases": [db.name for db in databases]} except (ValueError, requests.RequestException) as e: return _format_error_response(f"Failed to retrieve databases: {str(e)}") except Exception as e: return _format_error_response( f"Unexpected error while retrieving databases: {str(e)}" )
• td_mcp_server/api.py:15-25 (schema)

  Pydantic model Database used to parse and validate the database information returned from the Treasure Data API.
  class Database(BaseModel): """Model representing a Treasure Data database.""" name: str created_at: str updated_at: str count: int organization: str | None = None permission: str delete_protected: bool
• td_mcp_server/api.py:250-279 (helper)

  Helper method in TreasureDataClient that fetches the list of databases from the TD API endpoint '/database/list' and applies pagination.
  def get_databases( self, limit: int = 30, offset: int = 0, all_results: bool = False ) -> list[Database]: """ Retrieve a list of databases with pagination support. Args: limit: Maximum number of databases to retrieve (defaults to 30) offset: Index to start retrieving from (defaults to 0) all_results: If True, retrieves all databases ignoring limit and offset Returns: A list of Database objects Raises: requests.HTTPError: If the API returns an error response """ response = self._make_request("GET", "database/list") all_databases = [Database(**db) for db in response.get("databases", [])] if all_results: return all_databases else: end_index = ( offset + limit if offset + limit <= len(all_databases) else len(all_databases) ) return all_databases[offset:end_index]
• td_mcp_server/mcp_impl.py:127-148 (helper)

  Helper function to create the TreasureDataClient instance from environment variables, used by td_list_databases.
  def _create_client( include_workflow: bool = False, ) -> TreasureDataClient | dict[str, str]: """Create TreasureDataClient with environment credentials. Args: include_workflow: Whether to include workflow endpoint Returns: TreasureDataClient instance or error dict if API key missing """ api_key, endpoint, workflow_endpoint = _get_api_credentials() if not api_key: return _format_error_response("TD_API_KEY environment variable is not set") kwargs = {"api_key": api_key, "endpoint": endpoint} if include_workflow and workflow_endpoint: kwargs["workflow_endpoint"] = workflow_endpoint return TreasureDataClient(**kwargs)