list_workflows
Retrieve and filter workflow records from ServiceNow to manage automation processes, with options to limit results and apply status or name filters.
Instructions
List workflows from ServiceNow
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | Maximum number of records to return | |
| offset | No | Offset to start from | |
| active | No | Filter by active status | |
| name | No | Filter by name (contains) | |
| query | No | Additional query string |
Implementation Reference
- The handler function that executes the list_workflows tool logic, querying the ServiceNow wf_workflow table API with parameters and returning formatted results.
def list_workflows( auth_manager: AuthManager, server_config: ServerConfig, params: Dict[str, Any], ) -> Dict[str, Any]: """ List workflows from ServiceNow. Args: auth_manager: Authentication manager server_config: Server configuration params: Parameters for listing workflows Returns: Dictionary containing the list of workflows """ params = _unwrap_params(params, ListWorkflowsParams) # Get the correct auth_manager and server_config try: auth_manager, server_config = _get_auth_and_config(auth_manager, server_config) except ValueError as e: logger.error(f"Error getting auth and config: {e}") return {"error": str(e)} # Convert parameters to ServiceNow query format query_params = { "sysparm_limit": params.get("limit", 10), "sysparm_offset": params.get("offset", 0), } # Build query string query_parts = [] if params.get("active") is not None: query_parts.append(f"active={str(params['active']).lower()}") if params.get("name"): query_parts.append(f"nameLIKE{params['name']}") if params.get("query"): query_parts.append(params["query"]) if query_parts: query_params["sysparm_query"] = "^".join(query_parts) # Make the API request try: headers = auth_manager.get_headers() url = f"{server_config.instance_url}/api/now/table/wf_workflow" response = requests.get(url, headers=headers, params=query_params) response.raise_for_status() result = response.json() return { "workflows": result.get("result", []), "count": len(result.get("result", [])), "total": int(response.headers.get("X-Total-Count", 0)), } except requests.RequestException as e: logger.error(f"Error listing workflows: {e}") return {"error": str(e)} except Exception as e: logger.error(f"Unexpected error listing workflows: {e}") return {"error": str(e)} - Pydantic model defining the input schema (parameters) for the list_workflows tool.
class ListWorkflowsParams(BaseModel): """Parameters for listing workflows.""" limit: Optional[int] = Field(10, description="Maximum number of records to return") offset: Optional[int] = Field(0, description="Offset to start from") active: Optional[bool] = Field(None, description="Filter by active status") name: Optional[str] = Field(None, description="Filter by name (contains)") query: Optional[str] = Field(None, description="Additional query string") - src/servicenow_mcp/utils/tool_utils.py:490-496 (registration)Tool definition and registration entry for 'list_workflows' used by the MCP server to register the tool with its handler, schema, description, and serialization details.
"list_workflows": ( list_workflows_tool, ListWorkflowsParams, str, # Expects JSON string "List workflows from ServiceNow", "json", # Tool returns list/dict ), - Helper function to unwrap and validate input parameters using the Pydantic schema.
def _unwrap_params(params: Any, param_class: Type[T]) -> Dict[str, Any]: """ Unwrap parameters if they're wrapped in a Pydantic model. This helps handle cases where the parameters are passed as a model instead of a dict. """ if isinstance(params, dict): return params if isinstance(params, param_class): return params.dict(exclude_none=True) return params - Helper function to resolve and validate auth_manager and server_config arguments, handling potential order swaps.
def _get_auth_and_config( auth_manager_or_config: Union[AuthManager, ServerConfig], server_config_or_auth: Union[ServerConfig, AuthManager], ) -> tuple[AuthManager, ServerConfig]: """ Get the correct auth_manager and server_config objects. This function handles the case where the parameters might be swapped. Args: auth_manager_or_config: Either an AuthManager or a ServerConfig. server_config_or_auth: Either a ServerConfig or an AuthManager. Returns: tuple[AuthManager, ServerConfig]: The correct auth_manager and server_config. Raises: ValueError: If the parameters are not of the expected types. """ # Check if the parameters are in the correct order if isinstance(auth_manager_or_config, AuthManager) and isinstance(server_config_or_auth, ServerConfig): return auth_manager_or_config, server_config_or_auth # Check if the parameters are swapped if isinstance(auth_manager_or_config, ServerConfig) and isinstance(server_config_or_auth, AuthManager): return server_config_or_auth, auth_manager_or_config # If we get here, at least one of the parameters is not of the expected type if hasattr(auth_manager_or_config, "get_headers"): auth_manager = auth_manager_or_config elif hasattr(server_config_or_auth, "get_headers"): auth_manager = server_config_or_auth else: raise ValueError("Cannot find get_headers method in either auth_manager or server_config") if hasattr(auth_manager_or_config, "instance_url"): server_config = auth_manager_or_config elif hasattr(server_config_or_auth, "instance_url"): server_config = server_config_or_auth else: raise ValueError("Cannot find instance_url attribute in either auth_manager or server_config") return auth_manager, server_config