list_workflows
Retrieve and filter workflows from ServiceNow to manage automation processes, with options to limit results, filter by status, and search by name.
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 implements the core logic of the 'list_workflows' tool. It queries the ServiceNow 'wf_workflow' table API with parameters, handles authentication, builds queries for filtering, and returns a structured response with workflows list, count, and total.
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 BaseModel defining the input schema/parameters for the list_workflows tool, including optional fields for limit, offset, active filter, name filter, and custom query.
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:536-542 (registration)Tool registration entry in the central get_tool_definitions() function, mapping the tool name to its handler (list_workflows_tool), input schema (ListWorkflowsParams), return type hint, description, and serialization method.
"list_workflows": ( list_workflows_tool, ListWorkflowsParams, str, # Expects JSON string "List workflows from ServiceNow", "json", # Tool returns list/dict ), - src/servicenow_mcp/tools/__init__.py:78-91 (registration)Import statement in tools/__init__.py that exposes list_workflows for use in tool_utils.py and other modules.
from servicenow_mcp.tools.workflow_tools import ( activate_workflow, add_workflow_activity, create_workflow, deactivate_workflow, delete_workflow_activity, get_workflow_activities, get_workflow_details, list_workflow_versions, list_workflows, reorder_workflow_activities, update_workflow, update_workflow_activity, ) - Helper function used by list_workflows (and other tools) to flexibly resolve 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