Rossum MCP Server

Instructions

Implementation Reference

• mcp_server.py:155-158 (handler)

  The handler function for the 'rossum_get_workspaces' tool. It is decorated with @mcp.tool() for registration and delegates the implementation to _get_workspaces_impl().

  @mcp.tool()
  async def rossum_get_workspaces() -> List[Dict[str, Any]]:
      """Get all workspaces from the Rossum organization."""
      return await _get_workspaces_impl()

• mcp_server.py:116-119 (helper)

  Helper function that performs the actual API call to fetch all workspaces using the paginated request utility.

  async def _get_workspaces_impl() -> List[Dict[str, Any]]:
      """Get all workspaces, handling pagination."""
      # No summary_keys provided, so it will fetch all fields for each workspace.
      return await _rossum_unpaginated_request("GET", "/workspaces")

• mcp_server.py:56-89 (helper)

  Core utility function for making paginated requests to the Rossum API, used by _get_workspaces_impl() to handle pagination transparently.

  async def _rossum_unpaginated_request(method: str, path: str, summary_keys: Optional[List[str]] = None) -> List[Dict[str, Any]]:
      """Make a request to the Rossum API, handling pagination and summarization."""
      all_items = []
      current_path = path
  
      while current_path:
          try:
              page_data = await _rossum_request(method, current_path)
          except Exception as e:
              # Log or handle error, for now re-raising.
              # Consider logging: print(f"Error fetching Rossum data from {current_path}: {e}")
              raise
  
          if page_data and "results" in page_data and isinstance(page_data.get("results"), list):
              for item in page_data["results"]:
                  if summary_keys:
                      summary_item = {key: item.get(key) for key in summary_keys if key in item}
                      all_items.append(summary_item)
                  else:
                      all_items.append(item)  # Append the full item if no summary_keys
          
          pagination_info = page_data.get("pagination") if page_data else None
          next_page_url = pagination_info.get("next") if pagination_info else None
          
          if next_page_url:
              if next_page_url.startswith(ROSSUM_API_BASE):
                  current_path = next_page_url[len(ROSSUM_API_BASE):]
              else:
                  # Consider logging: print(f"Warning: next_page_url {next_page_url} does not match ROSSUM_API_BASE")
                  current_path = None 
          else:
              current_path = None
  
      return all_items

• mcp_server.py:28-53 (helper)

  Low-level HTTP request function to the Rossum API, handling errors and JSON parsing, used by the pagination helper.

  async def _rossum_request(method: str, path: str, **kwargs) -> Any:
      """Make a request to the Rossum API"""
      try:
          response = await client.request(
              method=method,
              url=f"{ROSSUM_API_BASE}{path}",
              headers=await get_rossum_headers(),
              **kwargs
          )
          response.raise_for_status()
          
          # Handle cases where Rossum might return empty body on success (e.g., 204)
          if response.status_code == 204:
              return None 
          return response.json()
      except httpx.HTTPStatusError as e:
          # Get error detail from response if possible
          error_detail = str(e)
          try:
              error_detail = e.response.json()
          except Exception:
              pass
          raise Exception(f"Rossum API error {e.response.status_code}: {error_detail}")
      except httpx.RequestError as e:
          raise Exception(f"Rossum API request failed: {str(e)}")