Rossum MCP Server

Instructions

Implementation Reference

• mcp_server.py:141-144 (handler)

  The main MCP tool handler for 'rossum_get_hooks', registered via @mcp.tool() decorator. It delegates to the _get_hooks_impl() helper for the actual API interaction.
  @mcp.tool() async def rossum_get_hooks() -> List[Dict[str, Any]]: """Get all serverless function hooks from the Rossum organization.""" return await _get_hooks_impl()
• mcp_server.py:107-110 (helper)

  Core implementation logic for fetching hooks: defines summary keys and calls the paginated request helper to retrieve hooks from Rossum API endpoint '/hooks'.
  async def _get_hooks_impl() -> List[Dict[str, Any]]: """Get all hooks with essential fields only, handling pagination.""" summary_keys = ["id", "name", "type", "url", "active", "events", "queues"] return await _rossum_unpaginated_request("GET", "/hooks", summary_keys=summary_keys)
• mcp_server.py:56-89 (helper)

  Reusable helper for making paginated requests to Rossum API, collecting all pages of results and optionally summarizing to specific keys. Used by _get_hooks_impl().
  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 helper to Rossum API, handles auth headers, errors, and JSON parsing. Called by paginated request 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)}")