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)}")