td_list_databases

Instructions

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.

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)