rossum_get_hooks

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