ServiceNow MCP Server

Instructions

Implementation Reference

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

  Main handler function that retrieves workflow activities from ServiceNow API. Automatically fetches the latest published version if none specified, queries wf_activity table ordered by order.

  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 model for input parameters: workflow_id (required str), version (optional str).

  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:511-517 (registration)

  Registers the 'get_workflow_activities' tool in the MCP server, linking the handler function, input schema, description, and output serialization as JSON.

  "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/utils/tool_utils.py:270-271 (registration)

  Imports and aliases the handler function for use in tool registration.

      get_workflow_activities as get_workflow_activities_tool,
  )

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

  Helper function to unwrap parameters from Pydantic model to dict, used in all workflow tools including get_workflow_activities.

  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