ServiceNow MCP Server

Instructions

Implementation Reference

• src/servicenow_mcp/tools/workflow_tools.py:360-448 (handler)

  Main handler function that implements the get_workflow_activities tool. It unwraps params, determines the workflow version (latest published if not specified), queries the wf_activity table via ServiceNow REST API ordered by 'order', and returns activities list with metadata.

  def get_workflow_activities(
      auth_manager: AuthManager,
      server_config: ServerConfig,
      params: Dict[str, Any],
  ) -> Dict[str, Any]:
      """
      Get activities for a specific workflow.
      
      Args:
          auth_manager: Authentication manager
          server_config: Server configuration
          params: Parameters for getting workflow activities
          
      Returns:
          Dict[str, Any]: List of workflow activities
      """
      # Unwrap parameters if needed
      params = _unwrap_params(params, GetWorkflowActivitiesParams)
      
      # 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)}
      
      workflow_id = params.get("workflow_id")
      if not workflow_id:
          return {"error": "Workflow ID is required"}
      
      version_id = params.get("version")
      
      # If no version specified, get the latest published version
      if not version_id:
          try:
              headers = auth_manager.get_headers()
              version_url = f"{server_config.instance_url}/api/now/table/wf_workflow_version"
              version_params = {
                  "sysparm_query": f"workflow={workflow_id}^published=true",
                  "sysparm_limit": 1,
                  "sysparm_orderby": "version DESC",
              }
              
              version_response = requests.get(version_url, headers=headers, params=version_params)
              version_response.raise_for_status()
              
              version_result = version_response.json()
              versions = version_result.get("result", [])
              
              if not versions:
                  return {
                      "error": f"No published versions found for workflow {workflow_id}",
                      "workflow_id": workflow_id,
                  }
              
              version_id = versions[0]["sys_id"]
          except requests.RequestException as e:
              logger.error(f"Error getting workflow version: {e}")
              return {"error": str(e)}
          except Exception as e:
              logger.error(f"Unexpected error getting workflow version: {e}")
              return {"error": str(e)}
      
      # Get activities for the version
      try:
          headers = auth_manager.get_headers()
          activities_url = f"{server_config.instance_url}/api/now/table/wf_activity"
          activities_params = {
              "sysparm_query": f"workflow_version={version_id}",
              "sysparm_orderby": "order",
          }
          
          activities_response = requests.get(activities_url, headers=headers, params=activities_params)
          activities_response.raise_for_status()
          
          activities_result = activities_response.json()
          return {
              "activities": activities_result.get("result", []),
              "count": len(activities_result.get("result", [])),
              "workflow_id": workflow_id,
              "version_id": version_id,
          }
      except requests.RequestException as e:
          logger.error(f"Error getting workflow activities: {e}")
          return {"error": str(e)}
      except Exception as e:
          logger.error(f"Unexpected error getting workflow activities: {e}")
          return {"error": str(e)}

• src/servicenow_mcp/tools/workflow_tools.py:48-53 (schema)

  Pydantic BaseModel defining input schema for the tool: requires workflow_id, optional version.

  class GetWorkflowActivitiesParams(BaseModel):
      """Parameters for getting workflow activities."""
      
      workflow_id: str = Field(..., description="Workflow ID or sys_id")
      version: Optional[str] = Field(None, description="Specific version to get activities for")

• src/servicenow_mcp/utils/tool_utils.py:557-563 (registration)

  MCP tool registration in get_tool_definitions(): maps 'get_workflow_activities' to the handler function, input schema, description, and serialization handling.

  "get_workflow_activities": (
      get_workflow_activities_tool,
      GetWorkflowActivitiesParams,
      str,  # Expects JSON string
      "Get activities for a specific workflow",
      "json",  # Tool returns list/dict
  ),

• src/servicenow_mcp/tools/__init__.py:78-91 (registration)

  Imports get_workflow_activities from workflow_tools.py into the tools package namespace, making it available for use and export via __all__.

  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,
  )

• src/servicenow_mcp/tools/workflow_tools.py:126-136 (helper)

  Helper function used by get_workflow_activities to normalize input params into a dict from Pydantic model.

  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