This is page 4 of 4. Use http://codebase.md/osomai/servicenow-mcp?lines=false&page={x} to view the full context.
# Directory Structure
```
├── .cursorrules
├── .DS_Store
├── .env.example
├── .gitignore
├── config
│ └── tool_packages.yaml
├── debug_workflow_api.py
├── Dockerfile
├── docs
│ ├── catalog_optimization_plan.md
│ ├── catalog_variables.md
│ ├── catalog.md
│ ├── change_management.md
│ ├── changeset_management.md
│ ├── incident_management.md
│ ├── knowledge_base.md
│ ├── user_management.md
│ └── workflow_management.md
├── examples
│ ├── catalog_integration_test.py
│ ├── catalog_optimization_example.py
│ ├── change_management_demo.py
│ ├── changeset_management_demo.py
│ ├── claude_catalog_demo.py
│ ├── claude_desktop_config.json
│ ├── claude_incident_demo.py
│ ├── debug_workflow_api.py
│ ├── wake_servicenow_instance.py
│ └── workflow_management_demo.py
├── LICENSE
├── prompts
│ └── add_servicenow_mcp_tool.md
├── pyproject.toml
├── README.md
├── scripts
│ ├── check_pdi_info.py
│ ├── check_pdi_status.py
│ ├── install_claude_desktop.sh
│ ├── setup_api_key.py
│ ├── setup_auth.py
│ ├── setup_oauth.py
│ ├── setup.sh
│ └── test_connection.py
├── src
│ ├── .DS_Store
│ └── servicenow_mcp
│ ├── __init__.py
│ ├── .DS_Store
│ ├── auth
│ │ ├── __init__.py
│ │ └── auth_manager.py
│ ├── cli.py
│ ├── server_sse.py
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── catalog_optimization.py
│ │ ├── catalog_tools.py
│ │ ├── catalog_variables.py
│ │ ├── change_tools.py
│ │ ├── changeset_tools.py
│ │ ├── epic_tools.py
│ │ ├── incident_tools.py
│ │ ├── knowledge_base.py
│ │ ├── project_tools.py
│ │ ├── script_include_tools.py
│ │ ├── scrum_task_tools.py
│ │ ├── story_tools.py
│ │ ├── user_tools.py
│ │ └── workflow_tools.py
│ └── utils
│ ├── __init__.py
│ ├── config.py
│ └── tool_utils.py
├── tests
│ ├── test_catalog_optimization.py
│ ├── test_catalog_resources.py
│ ├── test_catalog_tools.py
│ ├── test_catalog_variables.py
│ ├── test_change_tools.py
│ ├── test_changeset_resources.py
│ ├── test_changeset_tools.py
│ ├── test_config.py
│ ├── test_incident_tools.py
│ ├── test_knowledge_base.py
│ ├── test_script_include_resources.py
│ ├── test_script_include_tools.py
│ ├── test_server_catalog_optimization.py
│ ├── test_server_catalog.py
│ ├── test_server_workflow.py
│ ├── test_user_tools.py
│ ├── test_workflow_tools_direct.py
│ ├── test_workflow_tools_params.py
│ └── test_workflow_tools.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/src/servicenow_mcp/utils/tool_utils.py:
--------------------------------------------------------------------------------
```python
from typing import Any, Callable, Dict, Tuple, Type
# Import all necessary tool implementation functions and params models
# (This list needs to be kept complete and up-to-date)
from servicenow_mcp.tools.catalog_optimization import (
OptimizationRecommendationsParams,
UpdateCatalogItemParams,
)
from servicenow_mcp.tools.catalog_optimization import (
get_optimization_recommendations as get_optimization_recommendations_tool,
)
from servicenow_mcp.tools.catalog_optimization import (
update_catalog_item as update_catalog_item_tool,
)
from servicenow_mcp.tools.catalog_tools import (
CreateCatalogCategoryParams,
GetCatalogItemParams,
ListCatalogCategoriesParams,
ListCatalogItemsParams,
MoveCatalogItemsParams,
UpdateCatalogCategoryParams,
)
from servicenow_mcp.tools.catalog_tools import (
create_catalog_category as create_catalog_category_tool,
)
from servicenow_mcp.tools.catalog_tools import (
get_catalog_item as get_catalog_item_tool,
)
from servicenow_mcp.tools.catalog_tools import (
list_catalog_categories as list_catalog_categories_tool,
)
from servicenow_mcp.tools.catalog_tools import (
list_catalog_items as list_catalog_items_tool,
)
from servicenow_mcp.tools.catalog_tools import (
move_catalog_items as move_catalog_items_tool,
)
from servicenow_mcp.tools.catalog_tools import (
update_catalog_category as update_catalog_category_tool,
)
from servicenow_mcp.tools.catalog_variables import (
CreateCatalogItemVariableParams,
ListCatalogItemVariablesParams,
UpdateCatalogItemVariableParams,
)
from servicenow_mcp.tools.catalog_variables import (
create_catalog_item_variable as create_catalog_item_variable_tool,
)
from servicenow_mcp.tools.catalog_variables import (
list_catalog_item_variables as list_catalog_item_variables_tool,
)
from servicenow_mcp.tools.catalog_variables import (
update_catalog_item_variable as update_catalog_item_variable_tool,
)
from servicenow_mcp.tools.change_tools import (
AddChangeTaskParams,
ApproveChangeParams,
CreateChangeRequestParams,
GetChangeRequestDetailsParams,
ListChangeRequestsParams,
RejectChangeParams,
SubmitChangeForApprovalParams,
UpdateChangeRequestParams,
)
from servicenow_mcp.tools.change_tools import (
add_change_task as add_change_task_tool,
)
from servicenow_mcp.tools.change_tools import (
approve_change as approve_change_tool,
)
from servicenow_mcp.tools.change_tools import (
create_change_request as create_change_request_tool,
)
from servicenow_mcp.tools.change_tools import (
get_change_request_details as get_change_request_details_tool,
)
from servicenow_mcp.tools.change_tools import (
list_change_requests as list_change_requests_tool,
)
from servicenow_mcp.tools.change_tools import (
reject_change as reject_change_tool,
)
from servicenow_mcp.tools.change_tools import (
submit_change_for_approval as submit_change_for_approval_tool,
)
from servicenow_mcp.tools.change_tools import (
update_change_request as update_change_request_tool,
)
from servicenow_mcp.tools.changeset_tools import (
AddFileToChangesetParams,
CommitChangesetParams,
CreateChangesetParams,
GetChangesetDetailsParams,
ListChangesetsParams,
PublishChangesetParams,
UpdateChangesetParams,
)
from servicenow_mcp.tools.changeset_tools import (
add_file_to_changeset as add_file_to_changeset_tool,
)
from servicenow_mcp.tools.changeset_tools import (
commit_changeset as commit_changeset_tool,
)
from servicenow_mcp.tools.changeset_tools import (
create_changeset as create_changeset_tool,
)
from servicenow_mcp.tools.changeset_tools import (
get_changeset_details as get_changeset_details_tool,
)
from servicenow_mcp.tools.changeset_tools import (
list_changesets as list_changesets_tool,
)
from servicenow_mcp.tools.changeset_tools import (
publish_changeset as publish_changeset_tool,
)
from servicenow_mcp.tools.changeset_tools import (
update_changeset as update_changeset_tool,
)
from servicenow_mcp.tools.incident_tools import (
AddCommentParams,
CreateIncidentParams,
ListIncidentsParams,
ResolveIncidentParams,
UpdateIncidentParams,
GetIncidentByNumberParams,
)
from servicenow_mcp.tools.incident_tools import (
add_comment as add_comment_tool,
)
from servicenow_mcp.tools.incident_tools import (
create_incident as create_incident_tool,
)
from servicenow_mcp.tools.incident_tools import (
list_incidents as list_incidents_tool,
)
from servicenow_mcp.tools.incident_tools import (
resolve_incident as resolve_incident_tool,
)
from servicenow_mcp.tools.incident_tools import (
update_incident as update_incident_tool,
)
from servicenow_mcp.tools.incident_tools import (
get_incident_by_number as get_incident_by_number_tool,
)
from servicenow_mcp.tools.knowledge_base import (
CreateArticleParams,
CreateKnowledgeBaseParams,
GetArticleParams,
ListArticlesParams,
ListKnowledgeBasesParams,
PublishArticleParams,
UpdateArticleParams,
)
from servicenow_mcp.tools.knowledge_base import (
CreateCategoryParams as CreateKBCategoryParams, # Aliased
)
from servicenow_mcp.tools.knowledge_base import (
ListCategoriesParams as ListKBCategoriesParams, # Aliased
)
from servicenow_mcp.tools.knowledge_base import (
create_article as create_article_tool,
)
from servicenow_mcp.tools.knowledge_base import (
# create_category aliased in function call
create_knowledge_base as create_knowledge_base_tool,
)
from servicenow_mcp.tools.knowledge_base import (
get_article as get_article_tool,
)
from servicenow_mcp.tools.knowledge_base import (
list_articles as list_articles_tool,
)
from servicenow_mcp.tools.knowledge_base import (
# list_categories aliased in function call
list_knowledge_bases as list_knowledge_bases_tool,
)
from servicenow_mcp.tools.knowledge_base import (
publish_article as publish_article_tool,
)
from servicenow_mcp.tools.knowledge_base import (
update_article as update_article_tool,
)
from servicenow_mcp.tools.script_include_tools import (
CreateScriptIncludeParams,
DeleteScriptIncludeParams,
GetScriptIncludeParams,
ListScriptIncludesParams,
ScriptIncludeResponse,
UpdateScriptIncludeParams,
)
from servicenow_mcp.tools.script_include_tools import (
create_script_include as create_script_include_tool,
)
from servicenow_mcp.tools.script_include_tools import (
delete_script_include as delete_script_include_tool,
)
from servicenow_mcp.tools.script_include_tools import (
get_script_include as get_script_include_tool,
)
from servicenow_mcp.tools.script_include_tools import (
list_script_includes as list_script_includes_tool,
)
from servicenow_mcp.tools.script_include_tools import (
update_script_include as update_script_include_tool,
)
from servicenow_mcp.tools.user_tools import (
AddGroupMembersParams,
CreateGroupParams,
CreateUserParams,
GetUserParams,
ListGroupsParams,
ListUsersParams,
RemoveGroupMembersParams,
UpdateGroupParams,
UpdateUserParams,
)
from servicenow_mcp.tools.user_tools import (
add_group_members as add_group_members_tool,
)
from servicenow_mcp.tools.user_tools import (
create_group as create_group_tool,
)
from servicenow_mcp.tools.user_tools import (
create_user as create_user_tool,
)
from servicenow_mcp.tools.user_tools import (
get_user as get_user_tool,
)
from servicenow_mcp.tools.user_tools import (
list_groups as list_groups_tool,
)
from servicenow_mcp.tools.user_tools import (
list_users as list_users_tool,
)
from servicenow_mcp.tools.user_tools import (
remove_group_members as remove_group_members_tool,
)
from servicenow_mcp.tools.user_tools import (
update_group as update_group_tool,
)
from servicenow_mcp.tools.user_tools import (
update_user as update_user_tool,
)
from servicenow_mcp.tools.workflow_tools import (
ActivateWorkflowParams,
AddWorkflowActivityParams,
CreateWorkflowParams,
DeactivateWorkflowParams,
DeleteWorkflowActivityParams,
GetWorkflowActivitiesParams,
GetWorkflowDetailsParams,
ListWorkflowsParams,
ListWorkflowVersionsParams,
ReorderWorkflowActivitiesParams,
UpdateWorkflowActivityParams,
UpdateWorkflowParams,
)
from servicenow_mcp.tools.workflow_tools import (
activate_workflow as activate_workflow_tool,
)
from servicenow_mcp.tools.workflow_tools import (
add_workflow_activity as add_workflow_activity_tool,
)
from servicenow_mcp.tools.workflow_tools import (
create_workflow as create_workflow_tool,
)
from servicenow_mcp.tools.workflow_tools import (
deactivate_workflow as deactivate_workflow_tool,
)
from servicenow_mcp.tools.workflow_tools import (
delete_workflow_activity as delete_workflow_activity_tool,
)
from servicenow_mcp.tools.workflow_tools import (
get_workflow_activities as get_workflow_activities_tool,
)
from servicenow_mcp.tools.workflow_tools import (
get_workflow_details as get_workflow_details_tool,
)
from servicenow_mcp.tools.workflow_tools import (
list_workflow_versions as list_workflow_versions_tool,
)
from servicenow_mcp.tools.workflow_tools import (
list_workflows as list_workflows_tool,
)
from servicenow_mcp.tools.workflow_tools import (
reorder_workflow_activities as reorder_workflow_activities_tool,
)
from servicenow_mcp.tools.workflow_tools import (
update_workflow as update_workflow_tool,
)
from servicenow_mcp.tools.workflow_tools import (
update_workflow_activity as update_workflow_activity_tool,
)
from servicenow_mcp.tools.story_tools import (
CreateStoryParams,
UpdateStoryParams,
ListStoriesParams,
ListStoryDependenciesParams,
CreateStoryDependencyParams,
DeleteStoryDependencyParams,
)
from servicenow_mcp.tools.story_tools import (
create_story as create_story_tool,
update_story as update_story_tool,
list_stories as list_stories_tool,
list_story_dependencies as list_story_dependencies_tool,
create_story_dependency as create_story_dependency_tool,
delete_story_dependency as delete_story_dependency_tool,
)
from servicenow_mcp.tools.epic_tools import (
CreateEpicParams,
UpdateEpicParams,
ListEpicsParams,
)
from servicenow_mcp.tools.epic_tools import (
create_epic as create_epic_tool,
update_epic as update_epic_tool,
list_epics as list_epics_tool,
)
from servicenow_mcp.tools.scrum_task_tools import (
CreateScrumTaskParams,
UpdateScrumTaskParams,
ListScrumTasksParams,
)
from servicenow_mcp.tools.scrum_task_tools import (
create_scrum_task as create_scrum_task_tool,
update_scrum_task as update_scrum_task_tool,
list_scrum_tasks as list_scrum_tasks_tool,
)
from servicenow_mcp.tools.project_tools import (
CreateProjectParams,
UpdateProjectParams,
ListProjectsParams,
)
from servicenow_mcp.tools.project_tools import (
create_project as create_project_tool,
update_project as update_project_tool,
list_projects as list_projects_tool,
)
# Define a type alias for the Pydantic models or dataclasses used for params
ParamsModel = Type[Any] # Use Type[Any] for broader compatibility initially
# Define the structure of the tool definition tuple
ToolDefinition = Tuple[
Callable, # Implementation function
ParamsModel, # Pydantic model for parameters
Type, # Return type annotation (used for hints, not strictly enforced by low-level server)
str, # Description
str, # Serialization method ('str', 'json', 'dict', 'model_dump', etc.)
]
def get_tool_definitions(
create_kb_category_tool_impl: Callable, list_kb_categories_tool_impl: Callable
) -> Dict[str, ToolDefinition]:
"""
Returns a dictionary containing definitions for all available ServiceNow tools.
This centralizes the tool definitions for use in the server implementation.
Pass aliased functions for KB categories directly.
Returns:
Dict[str, ToolDefinition]: A dictionary mapping tool names to their definitions.
"""
tool_definitions: Dict[str, ToolDefinition] = {
# Incident Tools
"create_incident": (
create_incident_tool,
CreateIncidentParams,
str,
"Create a new incident in ServiceNow",
"str",
),
"update_incident": (
update_incident_tool,
UpdateIncidentParams,
str,
"Update an existing incident in ServiceNow",
"str",
),
"add_comment": (
add_comment_tool,
AddCommentParams,
str,
"Add a comment to an incident in ServiceNow",
"str",
),
"resolve_incident": (
resolve_incident_tool,
ResolveIncidentParams,
str,
"Resolve an incident in ServiceNow",
"str",
),
"list_incidents": (
list_incidents_tool,
ListIncidentsParams,
str, # Expects JSON string
"List incidents from ServiceNow",
"json", # Tool returns list/dict, needs JSON dump
),
"get_incident_by_number":(
get_incident_by_number_tool,
GetIncidentByNumberParams,
str,
"Incident details from ServiceNow",
"json_dict"
),
# Catalog Tools
"list_catalog_items": (
list_catalog_items_tool,
ListCatalogItemsParams,
str, # Expects JSON string
"List service catalog items.",
"json", # Tool returns list/dict
),
"get_catalog_item": (
get_catalog_item_tool,
GetCatalogItemParams,
str, # Expects JSON string
"Get a specific service catalog item.",
"json_dict", # Tool returns Pydantic model
),
"list_catalog_categories": (
list_catalog_categories_tool,
ListCatalogCategoriesParams,
str, # Expects JSON string
"List service catalog categories.",
"json", # Tool returns list/dict
),
"create_catalog_category": (
create_catalog_category_tool,
CreateCatalogCategoryParams,
str, # Expects JSON string
"Create a new service catalog category.",
"json_dict", # Tool returns Pydantic model
),
"update_catalog_category": (
update_catalog_category_tool,
UpdateCatalogCategoryParams,
str, # Expects JSON string
"Update an existing service catalog category.",
"json_dict", # Tool returns Pydantic model
),
"move_catalog_items": (
move_catalog_items_tool,
MoveCatalogItemsParams,
str, # Expects JSON string
"Move catalog items to a different category.",
"json_dict", # Tool returns Pydantic model
),
"get_optimization_recommendations": (
get_optimization_recommendations_tool,
OptimizationRecommendationsParams,
str, # Expects JSON string
"Get optimization recommendations for the service catalog.",
"json", # Tool returns list/dict
),
"update_catalog_item": (
update_catalog_item_tool,
UpdateCatalogItemParams,
str, # Expects JSON string
"Update a service catalog item.",
"json", # Tool returns Pydantic model
),
# Catalog Variables
"create_catalog_item_variable": (
create_catalog_item_variable_tool,
CreateCatalogItemVariableParams,
Dict[str, Any], # Expects dict
"Create a new catalog item variable",
"dict", # Tool returns Pydantic model
),
"list_catalog_item_variables": (
list_catalog_item_variables_tool,
ListCatalogItemVariablesParams,
Dict[str, Any], # Expects dict
"List catalog item variables",
"dict", # Tool returns Pydantic model
),
"update_catalog_item_variable": (
update_catalog_item_variable_tool,
UpdateCatalogItemVariableParams,
Dict[str, Any], # Expects dict
"Update a catalog item variable",
"dict", # Tool returns Pydantic model
),
# Change Management Tools
"create_change_request": (
create_change_request_tool,
CreateChangeRequestParams,
str,
"Create a new change request in ServiceNow",
"str",
),
"update_change_request": (
update_change_request_tool,
UpdateChangeRequestParams,
str,
"Update an existing change request in ServiceNow",
"str",
),
"list_change_requests": (
list_change_requests_tool,
ListChangeRequestsParams,
str, # Expects JSON string
"List change requests from ServiceNow",
"json", # Tool returns list/dict
),
"get_change_request_details": (
get_change_request_details_tool,
GetChangeRequestDetailsParams,
str, # Expects JSON string
"Get detailed information about a specific change request",
"json", # Tool returns list/dict
),
"add_change_task": (
add_change_task_tool,
AddChangeTaskParams,
str, # Expects JSON string
"Add a task to a change request",
"json_dict", # Tool returns Pydantic model
),
"submit_change_for_approval": (
submit_change_for_approval_tool,
SubmitChangeForApprovalParams,
str,
"Submit a change request for approval",
"str", # Tool returns simple message
),
"approve_change": (
approve_change_tool,
ApproveChangeParams,
str,
"Approve a change request",
"str", # Tool returns simple message
),
"reject_change": (
reject_change_tool,
RejectChangeParams,
str,
"Reject a change request",
"str", # Tool returns simple message
),
# Workflow Management Tools
"list_workflows": (
list_workflows_tool,
ListWorkflowsParams,
str, # Expects JSON string
"List workflows from ServiceNow",
"json", # Tool returns list/dict
),
"get_workflow_details": (
get_workflow_details_tool,
GetWorkflowDetailsParams,
str, # Expects JSON string
"Get detailed information about a specific workflow",
"json", # Tool returns list/dict
),
"list_workflow_versions": (
list_workflow_versions_tool,
ListWorkflowVersionsParams,
str, # Expects JSON string
"List workflow versions from ServiceNow",
"json", # Tool returns list/dict
),
"get_workflow_activities": (
get_workflow_activities_tool,
GetWorkflowActivitiesParams,
str, # Expects JSON string
"Get activities for a specific workflow",
"json", # Tool returns list/dict
),
"create_workflow": (
create_workflow_tool,
CreateWorkflowParams,
str, # Expects JSON string
"Create a new workflow in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"update_workflow": (
update_workflow_tool,
UpdateWorkflowParams,
str, # Expects JSON string
"Update an existing workflow in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"activate_workflow": (
activate_workflow_tool,
ActivateWorkflowParams,
str,
"Activate a workflow in ServiceNow",
"str", # Tool returns simple message
),
"deactivate_workflow": (
deactivate_workflow_tool,
DeactivateWorkflowParams,
str,
"Deactivate a workflow in ServiceNow",
"str", # Tool returns simple message
),
"add_workflow_activity": (
add_workflow_activity_tool,
AddWorkflowActivityParams,
str, # Expects JSON string
"Add a new activity to a workflow in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"update_workflow_activity": (
update_workflow_activity_tool,
UpdateWorkflowActivityParams,
str, # Expects JSON string
"Update an existing activity in a workflow",
"json_dict", # Tool returns Pydantic model
),
"delete_workflow_activity": (
delete_workflow_activity_tool,
DeleteWorkflowActivityParams,
str,
"Delete an activity from a workflow",
"str", # Tool returns simple message
),
"reorder_workflow_activities": (
reorder_workflow_activities_tool,
ReorderWorkflowActivitiesParams,
str,
"Reorder activities in a workflow",
"str", # Tool returns simple message
),
# Changeset Management Tools
"list_changesets": (
list_changesets_tool,
ListChangesetsParams,
str, # Expects JSON string
"List changesets from ServiceNow",
"json", # Tool returns list/dict
),
"get_changeset_details": (
get_changeset_details_tool,
GetChangesetDetailsParams,
str, # Expects JSON string
"Get detailed information about a specific changeset",
"json", # Tool returns list/dict
),
"create_changeset": (
create_changeset_tool,
CreateChangesetParams,
str, # Expects JSON string
"Create a new changeset in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"update_changeset": (
update_changeset_tool,
UpdateChangesetParams,
str, # Expects JSON string
"Update an existing changeset in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"commit_changeset": (
commit_changeset_tool,
CommitChangesetParams,
str,
"Commit a changeset in ServiceNow",
"str", # Tool returns simple message
),
"publish_changeset": (
publish_changeset_tool,
PublishChangesetParams,
str,
"Publish a changeset in ServiceNow",
"str", # Tool returns simple message
),
"add_file_to_changeset": (
add_file_to_changeset_tool,
AddFileToChangesetParams,
str,
"Add a file to a changeset in ServiceNow",
"str", # Tool returns simple message
),
# Script Include Tools
"list_script_includes": (
list_script_includes_tool,
ListScriptIncludesParams,
Dict[str, Any], # Expects dict
"List script includes from ServiceNow",
"raw_dict", # Tool returns raw dict
),
"get_script_include": (
get_script_include_tool,
GetScriptIncludeParams,
Dict[str, Any], # Expects dict
"Get a specific script include from ServiceNow",
"raw_dict", # Tool returns raw dict
),
"create_script_include": (
create_script_include_tool,
CreateScriptIncludeParams,
ScriptIncludeResponse, # Expects Pydantic model
"Create a new script include in ServiceNow",
"raw_pydantic", # Tool returns Pydantic model
),
"update_script_include": (
update_script_include_tool,
UpdateScriptIncludeParams,
ScriptIncludeResponse, # Expects Pydantic model
"Update an existing script include in ServiceNow",
"raw_pydantic", # Tool returns Pydantic model
),
"delete_script_include": (
delete_script_include_tool,
DeleteScriptIncludeParams,
str, # Expects JSON string
"Delete a script include in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
# Knowledge Base Tools
"create_knowledge_base": (
create_knowledge_base_tool,
CreateKnowledgeBaseParams,
str, # Expects JSON string
"Create a new knowledge base in ServiceNow",
"json_dict", # Tool returns Pydantic model
),
"list_knowledge_bases": (
list_knowledge_bases_tool,
ListKnowledgeBasesParams,
Dict[str, Any], # Expects dict
"List knowledge bases from ServiceNow",
"raw_dict", # Tool returns raw dict
),
# Use the passed-in implementations for aliased KB category tools
"create_category": (
create_kb_category_tool_impl, # Use passed function
CreateKBCategoryParams,
str, # Expects JSON string
"Create a new category in a knowledge base",
"json_dict", # Tool returns Pydantic model
),
"create_article": (
create_article_tool,
CreateArticleParams,
str, # Expects JSON string
"Create a new knowledge article",
"json_dict", # Tool returns Pydantic model
),
"update_article": (
update_article_tool,
UpdateArticleParams,
str, # Expects JSON string
"Update an existing knowledge article",
"json_dict", # Tool returns Pydantic model
),
"publish_article": (
publish_article_tool,
PublishArticleParams,
str, # Expects JSON string
"Publish a knowledge article",
"json_dict", # Tool returns Pydantic model
),
"list_articles": (
list_articles_tool,
ListArticlesParams,
Dict[str, Any], # Expects dict
"List knowledge articles",
"raw_dict", # Tool returns raw dict
),
"get_article": (
get_article_tool,
GetArticleParams,
Dict[str, Any], # Expects dict
"Get a specific knowledge article by ID",
"raw_dict", # Tool returns raw dict
),
# Use the passed-in implementations for aliased KB category tools
"list_categories": (
list_kb_categories_tool_impl, # Use passed function
ListKBCategoriesParams,
Dict[str, Any], # Expects dict
"List categories in a knowledge base",
"raw_dict", # Tool returns raw dict
),
# User Management Tools
"create_user": (
create_user_tool,
CreateUserParams,
Dict[str, Any], # Expects dict
"Create a new user in ServiceNow",
"raw_dict", # Tool returns raw dict
),
"update_user": (
update_user_tool,
UpdateUserParams,
Dict[str, Any], # Expects dict
"Update an existing user in ServiceNow",
"raw_dict",
),
"get_user": (
get_user_tool,
GetUserParams,
Dict[str, Any], # Expects dict
"Get a specific user in ServiceNow",
"raw_dict",
),
"list_users": (
list_users_tool,
ListUsersParams,
Dict[str, Any], # Expects dict
"List users in ServiceNow",
"raw_dict",
),
"create_group": (
create_group_tool,
CreateGroupParams,
Dict[str, Any], # Expects dict
"Create a new group in ServiceNow",
"raw_dict",
),
"update_group": (
update_group_tool,
UpdateGroupParams,
Dict[str, Any], # Expects dict
"Update an existing group in ServiceNow",
"raw_dict",
),
"add_group_members": (
add_group_members_tool,
AddGroupMembersParams,
Dict[str, Any], # Expects dict
"Add members to an existing group in ServiceNow",
"raw_dict",
),
"remove_group_members": (
remove_group_members_tool,
RemoveGroupMembersParams,
Dict[str, Any], # Expects dict
"Remove members from an existing group in ServiceNow",
"raw_dict",
),
"list_groups": (
list_groups_tool,
ListGroupsParams,
Dict[str, Any], # Expects dict
"List groups from ServiceNow with optional filtering",
"raw_dict",
),
# Story Management Tools
"create_story": (
create_story_tool,
CreateStoryParams,
str,
"Create a new story in ServiceNow",
"str",
),
"update_story": (
update_story_tool,
UpdateStoryParams,
str,
"Update an existing story in ServiceNow",
"str",
),
"list_stories": (
list_stories_tool,
ListStoriesParams,
str, # Expects JSON string
"List stories from ServiceNow",
"json", # Tool returns list/dict
),
"list_story_dependencies": (
list_story_dependencies_tool,
ListStoryDependenciesParams,
str, # Expects JSON string
"List story dependencies from ServiceNow",
"json", # Tool returns list/dict
),
"create_story_dependency": (
create_story_dependency_tool,
CreateStoryDependencyParams,
str,
"Create a dependency between two stories in ServiceNow",
"str",
),
"delete_story_dependency": (
delete_story_dependency_tool,
DeleteStoryDependencyParams,
str,
"Delete a story dependency in ServiceNow",
"str",
),
# Epic Management Tools
"create_epic": (
create_epic_tool,
CreateEpicParams,
str,
"Create a new epic in ServiceNow",
"str",
),
"update_epic": (
update_epic_tool,
UpdateEpicParams,
str,
"Update an existing epic in ServiceNow",
"str",
),
"list_epics": (
list_epics_tool,
ListEpicsParams,
str, # Expects JSON string
"List epics from ServiceNow",
"json", # Tool returns list/dict
),
# Scrum Task Management Tools
"create_scrum_task": (
create_scrum_task_tool,
CreateScrumTaskParams,
str,
"Create a new scrum task in ServiceNow",
"str",
),
"update_scrum_task": (
update_scrum_task_tool,
UpdateScrumTaskParams,
str,
"Update an existing scrum task in ServiceNow",
"str",
),
"list_scrum_tasks": (
list_scrum_tasks_tool,
ListScrumTasksParams,
str, # Expects JSON string
"List scrum tasks from ServiceNow",
"json", # Tool returns list/dict
),
# Project Management Tools
"create_project": (
create_project_tool,
CreateProjectParams,
str,
"Create a new project in ServiceNow",
"str",
),
"update_project": (
update_project_tool,
UpdateProjectParams,
str,
"Update an existing project in ServiceNow",
"str",
),
"list_projects": (
list_projects_tool,
ListProjectsParams,
str, # Expects JSON string
"List projects from ServiceNow",
"json", # Tool returns list/dict
),
}
return tool_definitions
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/change_tools.py:
--------------------------------------------------------------------------------
```python
"""
Change management tools for the ServiceNow MCP server.
This module provides tools for managing change requests in ServiceNow.
"""
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, Type, TypeVar
import requests
from pydantic import BaseModel, Field
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.utils.config import ServerConfig
logger = logging.getLogger(__name__)
# Type variable for Pydantic models
T = TypeVar('T', bound=BaseModel)
class CreateChangeRequestParams(BaseModel):
"""Parameters for creating a change request."""
short_description: str = Field(..., description="Short description of the change request")
description: Optional[str] = Field(None, description="Detailed description of the change request")
type: str = Field(..., description="Type of change (normal, standard, emergency)")
risk: Optional[str] = Field(None, description="Risk level of the change")
impact: Optional[str] = Field(None, description="Impact of the change")
category: Optional[str] = Field(None, description="Category of the change")
requested_by: Optional[str] = Field(None, description="User who requested the change")
assignment_group: Optional[str] = Field(None, description="Group assigned to the change")
start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
class UpdateChangeRequestParams(BaseModel):
"""Parameters for updating a change request."""
change_id: str = Field(..., description="Change request ID or sys_id")
short_description: Optional[str] = Field(None, description="Short description of the change request")
description: Optional[str] = Field(None, description="Detailed description of the change request")
state: Optional[str] = Field(None, description="State of the change request")
risk: Optional[str] = Field(None, description="Risk level of the change")
impact: Optional[str] = Field(None, description="Impact of the change")
category: Optional[str] = Field(None, description="Category of the change")
assignment_group: Optional[str] = Field(None, description="Group assigned to the change")
start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
work_notes: Optional[str] = Field(None, description="Work notes to add to the change request")
class ListChangeRequestsParams(BaseModel):
"""Parameters for listing change requests."""
limit: Optional[int] = Field(10, description="Maximum number of records to return")
offset: Optional[int] = Field(0, description="Offset to start from")
state: Optional[str] = Field(None, description="Filter by state")
type: Optional[str] = Field(None, description="Filter by type (normal, standard, emergency)")
category: Optional[str] = Field(None, description="Filter by category")
assignment_group: Optional[str] = Field(None, description="Filter by assignment group")
timeframe: Optional[str] = Field(None, description="Filter by timeframe (upcoming, in-progress, completed)")
query: Optional[str] = Field(None, description="Additional query string")
class GetChangeRequestDetailsParams(BaseModel):
"""Parameters for getting change request details."""
change_id: str = Field(..., description="Change request ID or sys_id")
class AddChangeTaskParams(BaseModel):
"""Parameters for adding a task to a change request."""
change_id: str = Field(..., description="Change request ID or sys_id")
short_description: str = Field(..., description="Short description of the task")
description: Optional[str] = Field(None, description="Detailed description of the task")
assigned_to: Optional[str] = Field(None, description="User assigned to the task")
planned_start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
planned_end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
class SubmitChangeForApprovalParams(BaseModel):
"""Parameters for submitting a change request for approval."""
change_id: str = Field(..., description="Change request ID or sys_id")
approval_comments: Optional[str] = Field(None, description="Comments for the approval request")
class ApproveChangeParams(BaseModel):
"""Parameters for approving a change request."""
change_id: str = Field(..., description="Change request ID or sys_id")
approver_id: Optional[str] = Field(None, description="ID of the approver")
approval_comments: Optional[str] = Field(None, description="Comments for the approval")
class RejectChangeParams(BaseModel):
"""Parameters for rejecting a change request."""
change_id: str = Field(..., description="Change request ID or sys_id")
approver_id: Optional[str] = Field(None, description="ID of the approver")
rejection_reason: str = Field(..., description="Reason for rejection")
def _unwrap_and_validate_params(params: Any, model_class: Type[T], required_fields: List[str] = None) -> Dict[str, Any]:
"""
Helper function to unwrap and validate parameters.
Args:
params: The parameters to unwrap and validate.
model_class: The Pydantic model class to validate against.
required_fields: List of required field names.
Returns:
A tuple of (success, result) where result is either the validated parameters or an error message.
"""
# Handle case where params might be wrapped in another dictionary
if isinstance(params, dict) and len(params) == 1 and "params" in params and isinstance(params["params"], dict):
logger.warning("Detected params wrapped in a 'params' key. Unwrapping...")
params = params["params"]
# Handle case where params might be a Pydantic model object
if not isinstance(params, dict):
try:
# Try to convert to dict if it's a Pydantic model
logger.warning("Params is not a dictionary. Attempting to convert...")
params = params.dict() if hasattr(params, "dict") else dict(params)
except Exception as e:
logger.error(f"Failed to convert params to dictionary: {e}")
return {
"success": False,
"message": f"Invalid parameters format. Expected a dictionary, got {type(params).__name__}",
}
# Validate required parameters are present
if required_fields:
for field in required_fields:
if field not in params:
return {
"success": False,
"message": f"Missing required parameter '{field}'",
}
try:
# Validate parameters against the model
validated_params = model_class(**params)
return {
"success": True,
"params": validated_params,
}
except Exception as e:
logger.error(f"Error validating parameters: {e}")
return {
"success": False,
"message": f"Error validating parameters: {str(e)}",
}
def _get_instance_url(auth_manager: AuthManager, server_config: ServerConfig) -> Optional[str]:
"""
Helper function to get the instance URL from either server_config or auth_manager.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
Returns:
The instance URL if found, None otherwise.
"""
if hasattr(server_config, 'instance_url'):
return server_config.instance_url
elif hasattr(auth_manager, 'instance_url'):
return auth_manager.instance_url
else:
logger.error("Cannot find instance_url in either server_config or auth_manager")
return None
def _get_headers(auth_manager: Any, server_config: Any) -> Optional[Dict[str, str]]:
"""
Helper function to get headers from either auth_manager or server_config.
Args:
auth_manager: The authentication manager or object passed as auth_manager.
server_config: The server configuration or object passed as server_config.
Returns:
The headers if found, None otherwise.
"""
# Try to get headers from auth_manager
if hasattr(auth_manager, 'get_headers'):
return auth_manager.get_headers()
# If auth_manager doesn't have get_headers, try server_config
if hasattr(server_config, 'get_headers'):
return server_config.get_headers()
# If neither has get_headers, check if auth_manager is actually a ServerConfig
# and server_config is actually an AuthManager (parameters swapped)
if hasattr(server_config, 'get_headers') and not hasattr(auth_manager, 'get_headers'):
return server_config.get_headers()
logger.error("Cannot find get_headers method in either auth_manager or server_config")
return None
def create_change_request(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Create a new change request in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for creating the change request.
Returns:
The created change request.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
CreateChangeRequestParams,
required_fields=["short_description", "type"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {
"short_description": validated_params.short_description,
"type": validated_params.type,
}
# Add optional fields if provided
if validated_params.description:
data["description"] = validated_params.description
if validated_params.risk:
data["risk"] = validated_params.risk
if validated_params.impact:
data["impact"] = validated_params.impact
if validated_params.category:
data["category"] = validated_params.category
if validated_params.requested_by:
data["requested_by"] = validated_params.requested_by
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.start_date:
data["start_date"] = validated_params.start_date
if validated_params.end_date:
data["end_date"] = validated_params.end_date
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Add Content-Type header
headers["Content-Type"] = "application/json"
# Make the API request
url = f"{instance_url}/api/now/table/change_request"
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Change request created successfully",
"change_request": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error creating change request: {e}")
return {
"success": False,
"message": f"Error creating change request: {str(e)}",
}
def update_change_request(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Update an existing change request in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for updating the change request.
Returns:
The updated change request.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
UpdateChangeRequestParams,
required_fields=["change_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {}
# Add fields if provided
if validated_params.short_description:
data["short_description"] = validated_params.short_description
if validated_params.description:
data["description"] = validated_params.description
if validated_params.state:
data["state"] = validated_params.state
if validated_params.risk:
data["risk"] = validated_params.risk
if validated_params.impact:
data["impact"] = validated_params.impact
if validated_params.category:
data["category"] = validated_params.category
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.start_date:
data["start_date"] = validated_params.start_date
if validated_params.end_date:
data["end_date"] = validated_params.end_date
if validated_params.work_notes:
data["work_notes"] = validated_params.work_notes
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Add Content-Type header
headers["Content-Type"] = "application/json"
# Make the API request
url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
try:
response = requests.put(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Change request updated successfully",
"change_request": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error updating change request: {e}")
return {
"success": False,
"message": f"Error updating change request: {str(e)}",
}
def list_change_requests(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
List change requests from ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for listing change requests.
Returns:
A list of change requests.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
ListChangeRequestsParams
)
if not result["success"]:
return result
validated_params = result["params"]
# Build the query
query_parts = []
if validated_params.state:
query_parts.append(f"state={validated_params.state}")
if validated_params.type:
query_parts.append(f"type={validated_params.type}")
if validated_params.category:
query_parts.append(f"category={validated_params.category}")
if validated_params.assignment_group:
query_parts.append(f"assignment_group={validated_params.assignment_group}")
# Handle timeframe filtering
if validated_params.timeframe:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if validated_params.timeframe == "upcoming":
query_parts.append(f"start_date>{now}")
elif validated_params.timeframe == "in-progress":
query_parts.append(f"start_date<{now}^end_date>{now}")
elif validated_params.timeframe == "completed":
query_parts.append(f"end_date<{now}")
# Add any additional query string
if validated_params.query:
query_parts.append(validated_params.query)
# Combine query parts
query = "^".join(query_parts) if query_parts else ""
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Make the API request
url = f"{instance_url}/api/now/table/change_request"
params = {
"sysparm_limit": validated_params.limit,
"sysparm_offset": validated_params.offset,
"sysparm_query": query,
"sysparm_display_value": "true",
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
result = response.json()
# Handle the case where result["result"] is a list
change_requests = result.get("result", [])
count = len(change_requests)
return {
"success": True,
"change_requests": change_requests,
"count": count,
"total": count, # Use count as total if total is not provided
}
except requests.exceptions.RequestException as e:
logger.error(f"Error listing change requests: {e}")
return {
"success": False,
"message": f"Error listing change requests: {str(e)}",
}
def get_change_request_details(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Get details of a change request from ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for getting change request details.
Returns:
The change request details.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
GetChangeRequestDetailsParams,
required_fields=["change_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Make the API request
url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
params = {
"sysparm_display_value": "true",
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
result = response.json()
# Get tasks associated with this change request
tasks_url = f"{instance_url}/api/now/table/change_task"
tasks_params = {
"sysparm_query": f"change_request={validated_params.change_id}",
"sysparm_display_value": "true",
}
tasks_response = requests.get(tasks_url, headers=headers, params=tasks_params)
tasks_response.raise_for_status()
tasks_result = tasks_response.json()
return {
"success": True,
"change_request": result["result"],
"tasks": tasks_result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error getting change request details: {e}")
return {
"success": False,
"message": f"Error getting change request details: {str(e)}",
}
def add_change_task(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Add a task to a change request in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for adding a change task.
Returns:
The created change task.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
AddChangeTaskParams,
required_fields=["change_id", "short_description"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {
"change_request": validated_params.change_id,
"short_description": validated_params.short_description,
}
# Add optional fields if provided
if validated_params.description:
data["description"] = validated_params.description
if validated_params.assigned_to:
data["assigned_to"] = validated_params.assigned_to
if validated_params.planned_start_date:
data["planned_start_date"] = validated_params.planned_start_date
if validated_params.planned_end_date:
data["planned_end_date"] = validated_params.planned_end_date
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Add Content-Type header
headers["Content-Type"] = "application/json"
# Make the API request
url = f"{instance_url}/api/now/table/change_task"
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Change task added successfully",
"change_task": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error adding change task: {e}")
return {
"success": False,
"message": f"Error adding change task: {str(e)}",
}
def submit_change_for_approval(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Submit a change request for approval in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for submitting a change request for approval.
Returns:
The result of the submission.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
SubmitChangeForApprovalParams,
required_fields=["change_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {
"state": "assess", # Set state to "assess" to submit for approval
}
# Add approval comments if provided
if validated_params.approval_comments:
data["work_notes"] = validated_params.approval_comments
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# Add Content-Type header
headers["Content-Type"] = "application/json"
# Make the API request
url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
try:
response = requests.patch(url, json=data, headers=headers)
response.raise_for_status()
# Now, create an approval request
approval_url = f"{instance_url}/api/now/table/sysapproval_approver"
approval_data = {
"document_id": validated_params.change_id,
"source_table": "change_request",
"state": "requested",
}
approval_response = requests.post(approval_url, json=approval_data, headers=headers)
approval_response.raise_for_status()
approval_result = approval_response.json()
return {
"success": True,
"message": "Change request submitted for approval successfully",
"approval": approval_result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error submitting change for approval: {e}")
return {
"success": False,
"message": f"Error submitting change for approval: {str(e)}",
}
def approve_change(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Approve a change request in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for approving a change request.
Returns:
The result of the approval.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
ApproveChangeParams,
required_fields=["change_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# First, find the approval record
approval_query_url = f"{instance_url}/api/now/table/sysapproval_approver"
query_params = {
"sysparm_query": f"document_id={validated_params.change_id}",
"sysparm_limit": 1,
}
try:
approval_response = requests.get(approval_query_url, headers=headers, params=query_params)
approval_response.raise_for_status()
approval_result = approval_response.json()
if not approval_result.get("result") or len(approval_result["result"]) == 0:
return {
"success": False,
"message": "No approval record found for this change request",
}
approval_id = approval_result["result"][0]["sys_id"]
# Now, update the approval record to approved
approval_update_url = f"{instance_url}/api/now/table/sysapproval_approver/{approval_id}"
headers["Content-Type"] = "application/json"
approval_data = {
"state": "approved",
}
if validated_params.approval_comments:
approval_data["comments"] = validated_params.approval_comments
approval_update_response = requests.patch(approval_update_url, json=approval_data, headers=headers)
approval_update_response.raise_for_status()
# Finally, update the change request state to "implement"
change_url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
change_data = {
"state": "implement", # This may vary depending on ServiceNow configuration
}
change_response = requests.patch(change_url, json=change_data, headers=headers)
change_response.raise_for_status()
return {
"success": True,
"message": "Change request approved successfully",
}
except requests.exceptions.RequestException as e:
logger.error(f"Error approving change: {e}")
return {
"success": False,
"message": f"Error approving change: {str(e)}",
}
def reject_change(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Reject a change request in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for rejecting a change request.
Returns:
The result of the rejection.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
RejectChangeParams,
required_fields=["change_id", "rejection_reason"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Get the instance URL
instance_url = _get_instance_url(auth_manager, server_config)
if not instance_url:
return {
"success": False,
"message": "Cannot find instance_url in either server_config or auth_manager",
}
# Get the headers
headers = _get_headers(auth_manager, server_config)
if not headers:
return {
"success": False,
"message": "Cannot find get_headers method in either auth_manager or server_config",
}
# First, find the approval record
approval_query_url = f"{instance_url}/api/now/table/sysapproval_approver"
query_params = {
"sysparm_query": f"document_id={validated_params.change_id}",
"sysparm_limit": 1,
}
try:
approval_response = requests.get(approval_query_url, headers=headers, params=query_params)
approval_response.raise_for_status()
approval_result = approval_response.json()
if not approval_result.get("result") or len(approval_result["result"]) == 0:
return {
"success": False,
"message": "No approval record found for this change request",
}
approval_id = approval_result["result"][0]["sys_id"]
# Now, update the approval record to rejected
approval_update_url = f"{instance_url}/api/now/table/sysapproval_approver/{approval_id}"
headers["Content-Type"] = "application/json"
approval_data = {
"state": "rejected",
"comments": validated_params.rejection_reason,
}
approval_update_response = requests.patch(approval_update_url, json=approval_data, headers=headers)
approval_update_response.raise_for_status()
# Finally, update the change request state to "canceled"
change_url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
change_data = {
"state": "canceled", # This may vary depending on ServiceNow configuration
"work_notes": f"Change request rejected: {validated_params.rejection_reason}",
}
change_response = requests.patch(change_url, json=change_data, headers=headers)
change_response.raise_for_status()
return {
"success": True,
"message": "Change request rejected successfully",
}
except requests.exceptions.RequestException as e:
logger.error(f"Error rejecting change: {e}")
return {
"success": False,
"message": f"Error rejecting change: {str(e)}",
}
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/workflow_tools.py:
--------------------------------------------------------------------------------
```python
"""
Workflow management tools for the ServiceNow MCP server.
This module provides tools for viewing and managing workflows in ServiceNow.
"""
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, Type, TypeVar, Union
import requests
from pydantic import BaseModel, Field
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.utils.config import ServerConfig
logger = logging.getLogger(__name__)
# Type variable for Pydantic models
T = TypeVar('T', bound=BaseModel)
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")
class GetWorkflowDetailsParams(BaseModel):
"""Parameters for getting workflow details."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
include_versions: Optional[bool] = Field(False, description="Include workflow versions")
class ListWorkflowVersionsParams(BaseModel):
"""Parameters for listing workflow versions."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
limit: Optional[int] = Field(10, description="Maximum number of records to return")
offset: Optional[int] = Field(0, description="Offset to start from")
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")
class CreateWorkflowParams(BaseModel):
"""Parameters for creating a new workflow."""
name: str = Field(..., description="Name of the workflow")
description: Optional[str] = Field(None, description="Description of the workflow")
table: Optional[str] = Field(None, description="Table the workflow applies to")
active: Optional[bool] = Field(True, description="Whether the workflow is active")
attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the workflow")
class UpdateWorkflowParams(BaseModel):
"""Parameters for updating a workflow."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
name: Optional[str] = Field(None, description="Name of the workflow")
description: Optional[str] = Field(None, description="Description of the workflow")
table: Optional[str] = Field(None, description="Table the workflow applies to")
active: Optional[bool] = Field(None, description="Whether the workflow is active")
attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the workflow")
class ActivateWorkflowParams(BaseModel):
"""Parameters for activating a workflow."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
class DeactivateWorkflowParams(BaseModel):
"""Parameters for deactivating a workflow."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
class AddWorkflowActivityParams(BaseModel):
"""Parameters for adding an activity to a workflow."""
workflow_version_id: str = Field(..., description="Workflow version ID")
name: str = Field(..., description="Name of the activity")
description: Optional[str] = Field(None, description="Description of the activity")
activity_type: str = Field(..., description="Type of activity (e.g., 'approval', 'task', 'notification')")
attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the activity")
class UpdateWorkflowActivityParams(BaseModel):
"""Parameters for updating a workflow activity."""
activity_id: str = Field(..., description="Activity ID or sys_id")
name: Optional[str] = Field(None, description="Name of the activity")
description: Optional[str] = Field(None, description="Description of the activity")
attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the activity")
class DeleteWorkflowActivityParams(BaseModel):
"""Parameters for deleting a workflow activity."""
activity_id: str = Field(..., description="Activity ID or sys_id")
class ReorderWorkflowActivitiesParams(BaseModel):
"""Parameters for reordering workflow activities."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
activity_ids: List[str] = Field(..., description="List of activity IDs in the desired order")
class DeleteWorkflowParams(BaseModel):
"""Parameters for deleting a workflow."""
workflow_id: str = Field(..., description="Workflow ID or sys_id")
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
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
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)}
def get_workflow_details(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Get detailed information about a specific workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for getting workflow details
Returns:
Dictionary containing the workflow details
"""
params = _unwrap_params(params, GetWorkflowDetailsParams)
# 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"}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
response = requests.get(url, headers=headers)
response.raise_for_status()
result = response.json()
return {
"workflow": result.get("result", {}),
}
except requests.RequestException as e:
logger.error(f"Error getting workflow details: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error getting workflow details: {e}")
return {"error": str(e)}
def list_workflow_versions(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
List versions of a specific workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for listing workflow versions
Returns:
Dict[str, Any]: List of workflow versions
"""
# Unwrap parameters if needed
params = _unwrap_params(params, ListWorkflowVersionsParams)
# 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"}
# Convert parameters to ServiceNow query format
query_params = {
"sysparm_query": f"workflow={workflow_id}",
"sysparm_limit": params.get("limit", 10),
"sysparm_offset": params.get("offset", 0),
}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow_version"
response = requests.get(url, headers=headers, params=query_params)
response.raise_for_status()
result = response.json()
return {
"versions": result.get("result", []),
"count": len(result.get("result", [])),
"total": int(response.headers.get("X-Total-Count", 0)),
"workflow_id": workflow_id,
}
except requests.RequestException as e:
logger.error(f"Error listing workflow versions: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error listing workflow versions: {e}")
return {"error": str(e)}
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)}
def create_workflow(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Create a new workflow in ServiceNow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for creating a workflow
Returns:
Dict[str, Any]: Created workflow details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, CreateWorkflowParams)
# 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)}
# Validate required parameters
if not params.get("name"):
return {"error": "Workflow name is required"}
# Prepare data for the API request
data = {
"name": params["name"],
}
if params.get("description"):
data["description"] = params["description"]
if params.get("table"):
data["table"] = params["table"]
if params.get("active") is not None:
data["active"] = str(params["active"]).lower()
if params.get("attributes"):
# Add any additional attributes
data.update(params["attributes"])
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow"
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"workflow": result.get("result", {}),
"message": "Workflow created successfully",
}
except requests.RequestException as e:
logger.error(f"Error creating workflow: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error creating workflow: {e}")
return {"error": str(e)}
def update_workflow(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Update an existing workflow in ServiceNow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for updating a workflow
Returns:
Dict[str, Any]: Updated workflow details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, UpdateWorkflowParams)
# 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"}
# Prepare data for the API request
data = {}
if params.get("name"):
data["name"] = params["name"]
if params.get("description") is not None:
data["description"] = params["description"]
if params.get("table"):
data["table"] = params["table"]
if params.get("active") is not None:
data["active"] = str(params["active"]).lower()
if params.get("attributes"):
# Add any additional attributes
data.update(params["attributes"])
if not data:
return {"error": "No update parameters provided"}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
response = requests.patch(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"workflow": result.get("result", {}),
"message": "Workflow updated successfully",
}
except requests.RequestException as e:
logger.error(f"Error updating workflow: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error updating workflow: {e}")
return {"error": str(e)}
def activate_workflow(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Activate a workflow in ServiceNow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for activating a workflow
Returns:
Dict[str, Any]: Activated workflow details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, ActivateWorkflowParams)
# 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"}
# Prepare data for the API request
data = {
"active": "true",
}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
response = requests.patch(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"workflow": result.get("result", {}),
"message": "Workflow activated successfully",
}
except requests.RequestException as e:
logger.error(f"Error activating workflow: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error activating workflow: {e}")
return {"error": str(e)}
def deactivate_workflow(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Deactivate a workflow in ServiceNow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for deactivating a workflow
Returns:
Dict[str, Any]: Deactivated workflow details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, DeactivateWorkflowParams)
# 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"}
# Prepare data for the API request
data = {
"active": "false",
}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
response = requests.patch(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"workflow": result.get("result", {}),
"message": "Workflow deactivated successfully",
}
except requests.RequestException as e:
logger.error(f"Error deactivating workflow: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error deactivating workflow: {e}")
return {"error": str(e)}
def add_workflow_activity(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Add a new activity to a workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for adding a workflow activity
Returns:
Dict[str, Any]: Added workflow activity details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, AddWorkflowActivityParams)
# 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)}
# Validate required parameters
workflow_version_id = params.get("workflow_version_id")
if not workflow_version_id:
return {"error": "Workflow version ID is required"}
activity_name = params.get("name")
if not activity_name:
return {"error": "Activity name is required"}
# Prepare data for the API request
data = {
"workflow_version": workflow_version_id,
"name": activity_name,
}
if params.get("description"):
data["description"] = params["description"]
if params.get("activity_type"):
data["activity_type"] = params["activity_type"]
if params.get("attributes"):
# Add any additional attributes
data.update(params["attributes"])
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_activity"
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"activity": result.get("result", {}),
"message": "Workflow activity added successfully",
}
except requests.RequestException as e:
logger.error(f"Error adding workflow activity: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error adding workflow activity: {e}")
return {"error": str(e)}
def update_workflow_activity(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Update an existing activity in a workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for updating a workflow activity
Returns:
Dict[str, Any]: Updated workflow activity details
"""
# Unwrap parameters if needed
params = _unwrap_params(params, UpdateWorkflowActivityParams)
# 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)}
activity_id = params.get("activity_id")
if not activity_id:
return {"error": "Activity ID is required"}
# Prepare data for the API request
data = {}
if params.get("name"):
data["name"] = params["name"]
if params.get("description") is not None:
data["description"] = params["description"]
if params.get("attributes"):
# Add any additional attributes
data.update(params["attributes"])
if not data:
return {"error": "No update parameters provided"}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
response = requests.patch(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
return {
"activity": result.get("result", {}),
"message": "Activity updated successfully",
}
except requests.RequestException as e:
logger.error(f"Error updating workflow activity: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error updating workflow activity: {e}")
return {"error": str(e)}
def delete_workflow_activity(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Delete an activity from a workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for deleting a workflow activity
Returns:
Dict[str, Any]: Result of the deletion operation
"""
# Unwrap parameters if needed
params = _unwrap_params(params, DeleteWorkflowActivityParams)
# 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)}
activity_id = params.get("activity_id")
if not activity_id:
return {"error": "Activity ID is required"}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
response = requests.delete(url, headers=headers)
response.raise_for_status()
return {
"message": "Activity deleted successfully",
"activity_id": activity_id,
}
except requests.RequestException as e:
logger.error(f"Error deleting workflow activity: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error deleting workflow activity: {e}")
return {"error": str(e)}
def reorder_workflow_activities(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Reorder activities in a workflow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for reordering workflow activities
Returns:
Dict[str, Any]: Result of the reordering operation
"""
# Unwrap parameters if needed
params = _unwrap_params(params, ReorderWorkflowActivitiesParams)
# 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"}
activity_ids = params.get("activity_ids")
if not activity_ids:
return {"error": "Activity IDs are required"}
# Make the API requests to update the order of each activity
try:
headers = auth_manager.get_headers()
results = []
for i, activity_id in enumerate(activity_ids):
# Calculate the new order value (100, 200, 300, etc.)
new_order = (i + 1) * 100
url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
data = {"order": new_order}
try:
response = requests.patch(url, headers=headers, json=data)
response.raise_for_status()
results.append({
"activity_id": activity_id,
"new_order": new_order,
"success": True,
})
except requests.RequestException as e:
logger.error(f"Error updating activity order: {e}")
results.append({
"activity_id": activity_id,
"error": str(e),
"success": False,
})
return {
"message": "Activities reordered",
"workflow_id": workflow_id,
"results": results,
}
except Exception as e:
logger.error(f"Unexpected error reordering workflow activities: {e}")
return {"error": str(e)}
def delete_workflow(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Delete a workflow from ServiceNow.
Args:
auth_manager: Authentication manager
server_config: Server configuration
params: Parameters for deleting a workflow
Returns:
Dict[str, Any]: Result of the deletion operation
"""
# Unwrap parameters if needed
params = _unwrap_params(params, DeleteWorkflowParams)
# 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"}
# Make the API request
try:
headers = auth_manager.get_headers()
url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
response = requests.delete(url, headers=headers)
response.raise_for_status()
return {
"message": f"Workflow {workflow_id} deleted successfully",
"workflow_id": workflow_id,
}
except requests.RequestException as e:
logger.error(f"Error deleting workflow: {e}")
return {"error": str(e)}
except Exception as e:
logger.error(f"Unexpected error deleting workflow: {e}")
return {"error": str(e)}
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/knowledge_base.py:
--------------------------------------------------------------------------------
```python
"""
Knowledge base tools for the ServiceNow MCP server.
This module provides tools for managing knowledge bases, categories, and articles in ServiceNow.
"""
import logging
from typing import Any, Dict, Optional
import requests
from pydantic import BaseModel, Field
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.utils.config import ServerConfig
logger = logging.getLogger(__name__)
class CreateKnowledgeBaseParams(BaseModel):
"""Parameters for creating a knowledge base."""
title: str = Field(..., description="Title of the knowledge base")
description: Optional[str] = Field(None, description="Description of the knowledge base")
owner: Optional[str] = Field(None, description="The specified admin user or group")
managers: Optional[str] = Field(None, description="Users who can manage this knowledge base")
publish_workflow: Optional[str] = Field("Knowledge - Instant Publish", description="Publication workflow")
retire_workflow: Optional[str] = Field("Knowledge - Instant Retire", description="Retirement workflow")
class ListKnowledgeBasesParams(BaseModel):
"""Parameters for listing knowledge bases."""
limit: int = Field(10, description="Maximum number of knowledge bases to return")
offset: int = Field(0, description="Offset for pagination")
active: Optional[bool] = Field(None, description="Filter by active status")
query: Optional[str] = Field(None, description="Search query for knowledge bases")
class CreateCategoryParams(BaseModel):
"""Parameters for creating a category in a knowledge base."""
title: str = Field(..., description="Title of the category")
description: Optional[str] = Field(None, description="Description of the category")
knowledge_base: str = Field(..., description="The knowledge base to create the category in")
parent_category: Optional[str] = Field(None, description="Parent category (if creating a subcategory). Sys_id refering to the parent category or sys_id of the parent table.")
parent_table: Optional[str] = Field(None, description="Parent table (if creating a subcategory). Sys_id refering to the table where the parent category is defined.")
active: bool = Field(True, description="Whether the category is active")
class CreateArticleParams(BaseModel):
"""Parameters for creating a knowledge article."""
title: str = Field(..., description="Title of the article")
text: str = Field(..., description="The main body text for the article. Field supports html formatting and wiki markup based on the article_type. HTML is the default.")
short_description: str = Field(..., description="Short description of the article")
knowledge_base: str = Field(..., description="The knowledge base to create the article in")
category: str = Field(..., description="Category for the article")
keywords: Optional[str] = Field(None, description="Keywords for search")
article_type: Optional[str] = Field("html", description="The type of article. Options are 'text' or 'wiki'. text lets the text field support html formatting. wiki lets the text field support wiki markup.")
class UpdateArticleParams(BaseModel):
"""Parameters for updating a knowledge article."""
article_id: str = Field(..., description="ID of the article to update")
title: Optional[str] = Field(None, description="Updated title of the article")
text: Optional[str] = Field(None, description="Updated main body text for the article. Field supports html formatting and wiki markup based on the article_type. HTML is the default.")
short_description: Optional[str] = Field(None, description="Updated short description")
category: Optional[str] = Field(None, description="Updated category for the article")
keywords: Optional[str] = Field(None, description="Updated keywords for search")
class PublishArticleParams(BaseModel):
"""Parameters for publishing a knowledge article."""
article_id: str = Field(..., description="ID of the article to publish")
workflow_state: Optional[str] = Field("published", description="The workflow state to set")
workflow_version: Optional[str] = Field(None, description="The workflow version to use")
class ListArticlesParams(BaseModel):
"""Parameters for listing knowledge articles."""
limit: int = Field(10, description="Maximum number of articles to return")
offset: int = Field(0, description="Offset for pagination")
knowledge_base: Optional[str] = Field(None, description="Filter by knowledge base")
category: Optional[str] = Field(None, description="Filter by category")
query: Optional[str] = Field(None, description="Search query for articles")
workflow_state: Optional[str] = Field(None, description="Filter by workflow state")
class GetArticleParams(BaseModel):
"""Parameters for getting a knowledge article."""
article_id: str = Field(..., description="ID of the article to get")
class KnowledgeBaseResponse(BaseModel):
"""Response from knowledge base operations."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
kb_id: Optional[str] = Field(None, description="ID of the affected knowledge base")
kb_name: Optional[str] = Field(None, description="Name of the affected knowledge base")
class CategoryResponse(BaseModel):
"""Response from category operations."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
category_id: Optional[str] = Field(None, description="ID of the affected category")
category_name: Optional[str] = Field(None, description="Name of the affected category")
class ArticleResponse(BaseModel):
"""Response from article operations."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
article_id: Optional[str] = Field(None, description="ID of the affected article")
article_title: Optional[str] = Field(None, description="Title of the affected article")
workflow_state: Optional[str] = Field(None, description="Current workflow state of the article")
class ListCategoriesParams(BaseModel):
"""Parameters for listing categories in a knowledge base."""
knowledge_base: Optional[str] = Field(None, description="Filter by knowledge base ID")
parent_category: Optional[str] = Field(None, description="Filter by parent category ID")
limit: int = Field(10, description="Maximum number of categories to return")
offset: int = Field(0, description="Offset for pagination")
active: Optional[bool] = Field(None, description="Filter by active status")
query: Optional[str] = Field(None, description="Search query for categories")
def create_knowledge_base(
config: ServerConfig,
auth_manager: AuthManager,
params: CreateKnowledgeBaseParams,
) -> KnowledgeBaseResponse:
"""
Create a new knowledge base in ServiceNow.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for creating the knowledge base.
Returns:
Response with the created knowledge base details.
"""
api_url = f"{config.api_url}/table/kb_knowledge_base"
# Build request data
data = {
"title": params.title,
}
if params.description:
data["description"] = params.description
if params.owner:
data["owner"] = params.owner
if params.managers:
data["kb_managers"] = params.managers
if params.publish_workflow:
data["workflow_publish"] = params.publish_workflow
if params.retire_workflow:
data["workflow_retire"] = params.retire_workflow
# Make request
try:
response = requests.post(
api_url,
json=data,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", {})
return KnowledgeBaseResponse(
success=True,
message="Knowledge base created successfully",
kb_id=result.get("sys_id"),
kb_name=result.get("title"),
)
except requests.RequestException as e:
logger.error(f"Failed to create knowledge base: {e}")
return KnowledgeBaseResponse(
success=False,
message=f"Failed to create knowledge base: {str(e)}",
)
def list_knowledge_bases(
config: ServerConfig,
auth_manager: AuthManager,
params: ListKnowledgeBasesParams,
) -> Dict[str, Any]:
"""
List knowledge bases with filtering options.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for listing knowledge bases.
Returns:
Dictionary with list of knowledge bases and metadata.
"""
api_url = f"{config.api_url}/table/kb_knowledge_base"
# Build query parameters
query_params = {
"sysparm_limit": params.limit,
"sysparm_offset": params.offset,
"sysparm_display_value": "true",
}
# Build query string
query_parts = []
if params.active is not None:
query_parts.append(f"active={str(params.active).lower()}")
if params.query:
query_parts.append(f"titleLIKE{params.query}^ORdescriptionLIKE{params.query}")
if query_parts:
query_params["sysparm_query"] = "^".join(query_parts)
# Make request
try:
response = requests.get(
api_url,
params=query_params,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
# Get the JSON response
json_response = response.json()
# Safely extract the result
if isinstance(json_response, dict) and "result" in json_response:
result = json_response.get("result", [])
else:
logger.error("Unexpected response format: %s", json_response)
return {
"success": False,
"message": "Unexpected response format",
"knowledge_bases": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
# Transform the results - create a simpler structure
knowledge_bases = []
# Handle either string or list
if isinstance(result, list):
for kb_item in result:
if not isinstance(kb_item, dict):
logger.warning("Skipping non-dictionary KB item: %s", kb_item)
continue
# Safely extract values
kb_id = kb_item.get("sys_id", "")
title = kb_item.get("title", "")
description = kb_item.get("description", "")
# Extract nested values safely
owner = ""
if isinstance(kb_item.get("owner"), dict):
owner = kb_item["owner"].get("display_value", "")
managers = ""
if isinstance(kb_item.get("kb_managers"), dict):
managers = kb_item["kb_managers"].get("display_value", "")
active = False
if kb_item.get("active") == "true":
active = True
created = kb_item.get("sys_created_on", "")
updated = kb_item.get("sys_updated_on", "")
knowledge_bases.append({
"id": kb_id,
"title": title,
"description": description,
"owner": owner,
"managers": managers,
"active": active,
"created": created,
"updated": updated,
})
else:
logger.warning("Result is not a list: %s", result)
return {
"success": True,
"message": f"Found {len(knowledge_bases)} knowledge bases",
"knowledge_bases": knowledge_bases,
"count": len(knowledge_bases),
"limit": params.limit,
"offset": params.offset,
}
except requests.RequestException as e:
logger.error(f"Failed to list knowledge bases: {e}")
return {
"success": False,
"message": f"Failed to list knowledge bases: {str(e)}",
"knowledge_bases": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
def create_category(
config: ServerConfig,
auth_manager: AuthManager,
params: CreateCategoryParams,
) -> CategoryResponse:
"""
Create a new category in a knowledge base.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for creating the category.
Returns:
Response with the created category details.
"""
api_url = f"{config.api_url}/table/kb_category"
# Build request data
data = {
"label": params.title,
"kb_knowledge_base": params.knowledge_base,
# Convert boolean to string "true"/"false" as ServiceNow expects
"active": str(params.active).lower(),
}
if params.description:
data["description"] = params.description
if params.parent_category:
data["parent"] = params.parent_category
if params.parent_table:
data["parent_table"] = params.parent_table
# Log the request data for debugging
logger.debug(f"Creating category with data: {data}")
# Make request
try:
response = requests.post(
api_url,
json=data,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", {})
logger.debug(f"Category creation response: {result}")
# Log the specific fields to check the knowledge base assignment
if "kb_knowledge_base" in result:
logger.debug(f"Knowledge base in response: {result['kb_knowledge_base']}")
# Log the active status
if "active" in result:
logger.debug(f"Active status in response: {result['active']}")
return CategoryResponse(
success=True,
message="Category created successfully",
category_id=result.get("sys_id"),
category_name=result.get("label"),
)
except requests.RequestException as e:
logger.error(f"Failed to create category: {e}")
return CategoryResponse(
success=False,
message=f"Failed to create category: {str(e)}",
)
def create_article(
config: ServerConfig,
auth_manager: AuthManager,
params: CreateArticleParams,
) -> ArticleResponse:
"""
Create a new knowledge article.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for creating the article.
Returns:
Response with the created article details.
"""
api_url = f"{config.api_url}/table/kb_knowledge"
# Build request data
data = {
"short_description": params.short_description,
"text": params.text,
"kb_knowledge_base": params.knowledge_base,
"kb_category": params.category,
"article_type": params.article_type,
}
if params.title:
data["short_description"] = params.title
if params.keywords:
data["keywords"] = params.keywords
# Make request
try:
response = requests.post(
api_url,
json=data,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", {})
return ArticleResponse(
success=True,
message="Article created successfully",
article_id=result.get("sys_id"),
article_title=result.get("short_description"),
workflow_state=result.get("workflow_state"),
)
except requests.RequestException as e:
logger.error(f"Failed to create article: {e}")
return ArticleResponse(
success=False,
message=f"Failed to create article: {str(e)}",
)
def update_article(
config: ServerConfig,
auth_manager: AuthManager,
params: UpdateArticleParams,
) -> ArticleResponse:
"""
Update an existing knowledge article.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for updating the article.
Returns:
Response with the updated article details.
"""
api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
# Build request data
data = {}
if params.title:
data["short_description"] = params.title
if params.text:
data["text"] = params.text
if params.short_description:
data["short_description"] = params.short_description
if params.category:
data["kb_category"] = params.category
if params.keywords:
data["keywords"] = params.keywords
# Make request
try:
response = requests.patch(
api_url,
json=data,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", {})
return ArticleResponse(
success=True,
message="Article updated successfully",
article_id=params.article_id,
article_title=result.get("short_description"),
workflow_state=result.get("workflow_state"),
)
except requests.RequestException as e:
logger.error(f"Failed to update article: {e}")
return ArticleResponse(
success=False,
message=f"Failed to update article: {str(e)}",
)
def publish_article(
config: ServerConfig,
auth_manager: AuthManager,
params: PublishArticleParams,
) -> ArticleResponse:
"""
Publish a knowledge article.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for publishing the article.
Returns:
Response with the published article details.
"""
api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
# Build request data
data = {
"workflow_state": params.workflow_state,
}
if params.workflow_version:
data["workflow_version"] = params.workflow_version
# Make request
try:
response = requests.patch(
api_url,
json=data,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", {})
return ArticleResponse(
success=True,
message="Article published successfully",
article_id=params.article_id,
article_title=result.get("short_description"),
workflow_state=result.get("workflow_state"),
)
except requests.RequestException as e:
logger.error(f"Failed to publish article: {e}")
return ArticleResponse(
success=False,
message=f"Failed to publish article: {str(e)}",
)
def list_articles(
config: ServerConfig,
auth_manager: AuthManager,
params: ListArticlesParams,
) -> Dict[str, Any]:
"""
List knowledge articles with filtering options.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for listing articles.
Returns:
Dictionary with list of articles and metadata.
"""
api_url = f"{config.api_url}/table/kb_knowledge"
# Build query parameters
query_params = {
"sysparm_limit": params.limit,
"sysparm_offset": params.offset,
"sysparm_display_value": "all",
}
# Build query string
query_parts = []
if params.knowledge_base:
query_parts.append(f"kb_knowledge_base.sys_id={params.knowledge_base}")
if params.category:
query_parts.append(f"kb_category.sys_id={params.category}")
if params.workflow_state:
query_parts.append(f"workflow_state={params.workflow_state}")
if params.query:
query_parts.append(f"short_descriptionLIKE{params.query}^ORtextLIKE{params.query}")
if query_parts:
query_string = "^".join(query_parts)
logger.debug(f"Constructed article query string: {query_string}")
query_params["sysparm_query"] = query_string
# Log the query parameters for debugging
logger.debug(f"Listing articles with query params: {query_params}")
# Make request
try:
response = requests.get(
api_url,
params=query_params,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
# Get the JSON response
json_response = response.json()
logger.debug(f"Article listing raw response: {json_response}")
# Safely extract the result
if isinstance(json_response, dict) and "result" in json_response:
result = json_response.get("result", [])
else:
logger.error("Unexpected response format: %s", json_response)
return {
"success": False,
"message": f"Unexpected response format",
"articles": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
# Transform the results
articles = []
# Handle either string or list
if isinstance(result, list):
for article_item in result:
if not isinstance(article_item, dict):
logger.warning("Skipping non-dictionary article item: %s", article_item)
continue
# Safely extract values
article_id = article_item.get("sys_id", "")
title = article_item.get("short_description", "")
# Extract nested values safely
knowledge_base = ""
if isinstance(article_item.get("kb_knowledge_base"), dict):
knowledge_base = article_item["kb_knowledge_base"].get("display_value", "")
category = ""
if isinstance(article_item.get("kb_category"), dict):
category = article_item["kb_category"].get("display_value", "")
workflow_state = ""
if isinstance(article_item.get("workflow_state"), dict):
workflow_state = article_item["workflow_state"].get("display_value", "")
created = article_item.get("sys_created_on", "")
updated = article_item.get("sys_updated_on", "")
articles.append({
"id": article_id,
"title": title,
"knowledge_base": knowledge_base,
"category": category,
"workflow_state": workflow_state,
"created": created,
"updated": updated,
})
else:
logger.warning("Result is not a list: %s", result)
return {
"success": True,
"message": f"Found {len(articles)} articles",
"articles": articles,
"count": len(articles),
"limit": params.limit,
"offset": params.offset,
}
except requests.RequestException as e:
logger.error(f"Failed to list articles: {e}")
return {
"success": False,
"message": f"Failed to list articles: {str(e)}",
"articles": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
def get_article(
config: ServerConfig,
auth_manager: AuthManager,
params: GetArticleParams,
) -> Dict[str, Any]:
"""
Get a specific knowledge article by ID.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for getting the article.
Returns:
Dictionary with article details.
"""
api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
# Build query parameters
query_params = {
"sysparm_display_value": "true",
}
# Make request
try:
response = requests.get(
api_url,
params=query_params,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
# Get the JSON response
json_response = response.json()
# Safely extract the result
if isinstance(json_response, dict) and "result" in json_response:
result = json_response.get("result", {})
else:
logger.error("Unexpected response format: %s", json_response)
return {
"success": False,
"message": "Unexpected response format",
}
if not result or not isinstance(result, dict):
return {
"success": False,
"message": f"Article with ID {params.article_id} not found",
}
# Extract values safely
article_id = result.get("sys_id", "")
title = result.get("short_description", "")
text = result.get("text", "")
# Extract nested values safely
knowledge_base = ""
if isinstance(result.get("kb_knowledge_base"), dict):
knowledge_base = result["kb_knowledge_base"].get("display_value", "")
category = ""
if isinstance(result.get("kb_category"), dict):
category = result["kb_category"].get("display_value", "")
workflow_state = ""
if isinstance(result.get("workflow_state"), dict):
workflow_state = result["workflow_state"].get("display_value", "")
author = ""
if isinstance(result.get("author"), dict):
author = result["author"].get("display_value", "")
keywords = result.get("keywords", "")
article_type = result.get("article_type", "")
views = result.get("view_count", "0")
created = result.get("sys_created_on", "")
updated = result.get("sys_updated_on", "")
article = {
"id": article_id,
"title": title,
"text": text,
"knowledge_base": knowledge_base,
"category": category,
"workflow_state": workflow_state,
"created": created,
"updated": updated,
"author": author,
"keywords": keywords,
"article_type": article_type,
"views": views,
}
return {
"success": True,
"message": "Article retrieved successfully",
"article": article,
}
except requests.RequestException as e:
logger.error(f"Failed to get article: {e}")
return {
"success": False,
"message": f"Failed to get article: {str(e)}",
}
def list_categories(
config: ServerConfig,
auth_manager: AuthManager,
params: ListCategoriesParams,
) -> Dict[str, Any]:
"""
List categories in a knowledge base.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for listing categories.
Returns:
Dictionary with list of categories and metadata.
"""
api_url = f"{config.api_url}/table/kb_category"
# Build query parameters
query_params = {
"sysparm_limit": params.limit,
"sysparm_offset": params.offset,
"sysparm_display_value": "all",
}
# Build query string
query_parts = []
if params.knowledge_base:
# Try different query format to ensure we match by sys_id value
query_parts.append(f"kb_knowledge_base.sys_id={params.knowledge_base}")
if params.parent_category:
query_parts.append(f"parent.sys_id={params.parent_category}")
if params.active is not None:
query_parts.append(f"active={str(params.active).lower()}")
if params.query:
query_parts.append(f"labelLIKE{params.query}^ORdescriptionLIKE{params.query}")
if query_parts:
query_string = "^".join(query_parts)
logger.debug(f"Constructed query string: {query_string}")
query_params["sysparm_query"] = query_string
# Log the query parameters for debugging
logger.debug(f"Listing categories with query params: {query_params}")
# Make request
try:
response = requests.get(
api_url,
params=query_params,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
# Get the JSON response
json_response = response.json()
# Safely extract the result
if isinstance(json_response, dict) and "result" in json_response:
result = json_response.get("result", [])
else:
logger.error("Unexpected response format: %s", json_response)
return {
"success": False,
"message": "Unexpected response format",
"categories": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
# Transform the results
categories = []
# Handle either string or list
if isinstance(result, list):
for category_item in result:
if not isinstance(category_item, dict):
logger.warning("Skipping non-dictionary category item: %s", category_item)
continue
# Safely extract values
category_id = category_item.get("sys_id", "")
title = category_item.get("label", "")
description = category_item.get("description", "")
# Extract knowledge base - handle both dictionary and string cases
knowledge_base = ""
kb_field = category_item.get("kb_knowledge_base")
if isinstance(kb_field, dict):
knowledge_base = kb_field.get("display_value", "")
elif isinstance(kb_field, str):
knowledge_base = kb_field
# Also check if kb_knowledge_base is missing but there's a separate value field
elif "kb_knowledge_base_value" in category_item:
knowledge_base = category_item.get("kb_knowledge_base_value", "")
elif "kb_knowledge_base.display_value" in category_item:
knowledge_base = category_item.get("kb_knowledge_base.display_value", "")
# Extract parent category - handle both dictionary and string cases
parent = ""
parent_field = category_item.get("parent")
if isinstance(parent_field, dict):
parent = parent_field.get("display_value", "")
elif isinstance(parent_field, str):
parent = parent_field
# Also check alternative field names
elif "parent_value" in category_item:
parent = category_item.get("parent_value", "")
elif "parent.display_value" in category_item:
parent = category_item.get("parent.display_value", "")
# Convert active to boolean - handle string or boolean types
active_field = category_item.get("active")
if isinstance(active_field, str):
active = active_field.lower() == "true"
elif isinstance(active_field, bool):
active = active_field
else:
active = False
created = category_item.get("sys_created_on", "")
updated = category_item.get("sys_updated_on", "")
categories.append({
"id": category_id,
"title": title,
"description": description,
"knowledge_base": knowledge_base,
"parent_category": parent,
"active": active,
"created": created,
"updated": updated,
})
# Log for debugging purposes
logger.debug(f"Processed category: {title}, KB: {knowledge_base}, Parent: {parent}")
else:
logger.warning("Result is not a list: %s", result)
return {
"success": True,
"message": f"Found {len(categories)} categories",
"categories": categories,
"count": len(categories),
"limit": params.limit,
"offset": params.offset,
}
except requests.RequestException as e:
logger.error(f"Failed to list categories: {e}")
return {
"success": False,
"message": f"Failed to list categories: {str(e)}",
"categories": [],
"count": 0,
"limit": params.limit,
"offset": params.offset,
}
```