This is page 2 of 4. Use http://codebase.md/osomai/servicenow-mcp?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
--------------------------------------------------------------------------------
/scripts/setup_oauth.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
ServiceNow OAuth Setup Script
This script helps set up and test OAuth authentication with ServiceNow.
It will:
1. Get an OAuth token using client credentials
2. Test the token with a simple API call
3. Update the .env file with the OAuth configuration
Usage:
python scripts/setup_oauth.py
"""
import os
import sys
import json
import requests
import base64
from pathlib import Path
from dotenv import load_dotenv
# Add the project root to the Python path
sys.path.insert(0, str(Path(__file__).parent.parent))
def setup_oauth():
# Load environment variables
load_dotenv()
# Get ServiceNow instance URL
instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
if not instance_url or instance_url == "https://your-instance.service-now.com":
instance_url = input("Enter your ServiceNow instance URL (e.g., https://dev296866.service-now.com): ")
print("\n=== ServiceNow OAuth Setup ===")
print("This script will help you set up OAuth authentication for your ServiceNow instance.")
print("You'll need to create an OAuth API client in ServiceNow first.")
print("\nTo create an OAuth API client:")
print("1. Log in to your ServiceNow instance")
print("2. Navigate to System OAuth > Application Registry")
print("3. Click 'New'")
print("4. Select 'Create an OAuth API endpoint for external clients'")
print("5. Fill in the required fields:")
print(" - Name: MCP Client (or any name you prefer)")
print(" - Redirect URL: http://localhost (for testing)")
print(" - Active: Checked")
print(" - Refresh Token Lifespan: 8 hours (or your preference)")
print(" - Access Token Lifespan: 30 minutes (or your preference)")
print("6. Save the application and note down the Client ID and Client Secret")
print("7. Go to the 'OAuth Scopes' related list and add appropriate scopes (e.g., 'admin')")
# Get OAuth credentials
client_id = input("\nEnter your Client ID: ")
client_secret = input("Enter your Client Secret: ")
# Get username and password for resource owner grant
username = os.getenv("SERVICENOW_USERNAME")
password = os.getenv("SERVICENOW_PASSWORD")
if not username or username == "your-username":
username = input("Enter your ServiceNow username: ")
if not password or password == "your-password":
password = input("Enter your ServiceNow password: ")
# Set token URL
token_url = f"{instance_url}/oauth_token.do"
print(f"\nTesting OAuth connection to {instance_url}...")
print("Trying different OAuth grant types...")
# Try different OAuth grant types
access_token = None
# 1. Try client credentials grant
try:
print("\nAttempting client_credentials grant...")
# Create authorization header
auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
token_response = requests.post(
token_url,
headers={
"Authorization": f"Basic {auth_header}",
"Content-Type": "application/x-www-form-urlencoded"
},
data={
"grant_type": "client_credentials"
}
)
if token_response.status_code == 200:
token_data = token_response.json()
access_token = token_data.get("access_token")
print("✅ Successfully obtained OAuth token using client_credentials grant!")
else:
print(f"❌ Failed with client_credentials grant: {token_response.status_code}")
print(f"Response: {token_response.text}")
except Exception as e:
print(f"❌ Error with client_credentials grant: {e}")
# 2. Try password grant if client credentials failed
if not access_token:
try:
print("\nAttempting password grant...")
token_response = requests.post(
token_url,
headers={
"Authorization": f"Basic {auth_header}",
"Content-Type": "application/x-www-form-urlencoded"
},
data={
"grant_type": "password",
"username": username,
"password": password
}
)
if token_response.status_code == 200:
token_data = token_response.json()
access_token = token_data.get("access_token")
print("✅ Successfully obtained OAuth token using password grant!")
else:
print(f"❌ Failed with password grant: {token_response.status_code}")
print(f"Response: {token_response.text}")
except Exception as e:
print(f"❌ Error with password grant: {e}")
# If we have a token, test it
if access_token:
# Test the token with a simple API call
test_url = f"{instance_url}/api/now/table/incident?sysparm_limit=1"
test_response = requests.get(
test_url,
headers={
"Authorization": f"Bearer {access_token}",
"Accept": "application/json"
}
)
if test_response.status_code == 200:
print("✅ Successfully tested OAuth token with API call!")
data = test_response.json()
print(f"Retrieved {len(data.get('result', []))} incident(s)")
# Update .env file
update_env = input("\nDo you want to update your .env file with these OAuth credentials? (y/n): ")
if update_env.lower() == 'y':
env_path = Path(__file__).parent.parent / '.env'
with open(env_path, 'r') as f:
env_lines = f.readlines()
# Helper to update or insert a key
def set_env_var(lines, key, value):
found = False
for i, line in enumerate(lines):
if line.strip().startswith(f'{key}=') or line.strip().startswith(f'#{key}='):
lines[i] = f'{key}={value}\n'
found = True
break
if not found:
lines.append(f'{key}={value}\n')
return lines
# Update or insert OAuth configuration
env_lines = set_env_var(env_lines, 'SERVICENOW_AUTH_TYPE', 'oauth')
env_lines = set_env_var(env_lines, 'SERVICENOW_CLIENT_ID', client_id)
env_lines = set_env_var(env_lines, 'SERVICENOW_CLIENT_SECRET', client_secret)
env_lines = set_env_var(env_lines, 'SERVICENOW_TOKEN_URL', token_url)
env_lines = set_env_var(env_lines, 'SERVICENOW_USERNAME', username)
env_lines = set_env_var(env_lines, 'SERVICENOW_PASSWORD', password)
with open(env_path, 'w') as f:
f.writelines(env_lines)
print("✅ Updated .env file with OAuth configuration!")
print("\nYou can now use OAuth authentication with the ServiceNow MCP server.")
print("To test it, run: python scripts/test_connection.py")
return True
else:
print(f"❌ Failed to test OAuth token with API call: {test_response.status_code}")
print(f"Response: {test_response.text}")
return False
else:
print("\n❌ Failed to obtain OAuth token with any grant type.")
print("\nPossible issues:")
print("1. The OAuth client may not have the correct scopes")
print("2. The client ID or client secret may be incorrect")
print("3. The OAuth client may not be active")
print("4. The username/password may be incorrect")
print("\nPlease check your ServiceNow instance OAuth configuration and try again.")
return False
if __name__ == "__main__":
setup_oauth()
```
--------------------------------------------------------------------------------
/examples/workflow_management_demo.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Workflow Management Demo
This script demonstrates how to use the ServiceNow MCP workflow management tools
to view, create, and modify workflows in ServiceNow.
"""
import os
import sys
import json
from datetime import datetime
from dotenv import load_dotenv
# Add the parent directory to the path so we can import the package
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
from servicenow_mcp.tools.workflow_tools import (
list_workflows,
get_workflow_details,
list_workflow_versions,
get_workflow_activities,
create_workflow,
update_workflow,
activate_workflow,
deactivate_workflow,
add_workflow_activity,
update_workflow_activity,
delete_workflow_activity,
reorder_workflow_activities,
)
def print_json(data):
"""Print JSON data in a readable format."""
print(json.dumps(data, indent=2))
def main():
"""Main function to demonstrate workflow management tools."""
# Load environment variables from .env file
load_dotenv()
# Get ServiceNow credentials from environment variables
instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
username = os.getenv("SERVICENOW_USERNAME")
password = os.getenv("SERVICENOW_PASSWORD")
if not all([instance_url, username, password]):
print("Error: Missing required environment variables.")
print("Please set SERVICENOW_INSTANCE_URL, SERVICENOW_USERNAME, and SERVICENOW_PASSWORD.")
sys.exit(1)
# Create authentication configuration
auth_config = AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(username=username, password=password),
)
# Create server configuration
server_config = ServerConfig(
instance_url=instance_url,
auth=auth_config,
)
# Create authentication manager
auth_manager = AuthManager(auth_config)
print("ServiceNow Workflow Management Demo")
print("===================================")
print(f"Instance URL: {instance_url}")
print(f"Username: {username}")
print()
# List active workflows
print("Listing active workflows...")
workflows_result = list_workflows(
auth_manager,
server_config,
{
"limit": 5,
"active": True,
},
)
print_json(workflows_result)
print()
# Check if we have any workflows
if workflows_result.get("count", 0) == 0:
print("No active workflows found. Creating a new workflow...")
# Create a new workflow
new_workflow_result = create_workflow(
auth_manager,
server_config,
{
"name": f"Demo Workflow {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
"description": "A demo workflow created by the ServiceNow MCP workflow management demo",
"table": "incident",
"active": True,
},
)
print_json(new_workflow_result)
print()
if "error" in new_workflow_result:
print("Error creating workflow. Exiting.")
sys.exit(1)
workflow_id = new_workflow_result["workflow"]["sys_id"]
else:
# Use the first workflow from the list
workflow_id = workflows_result["workflows"][0]["sys_id"]
print(f"Using workflow with ID: {workflow_id}")
print()
# Get workflow details
print("Getting workflow details...")
workflow_details = get_workflow_details(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(workflow_details)
print()
# List workflow versions
print("Listing workflow versions...")
versions_result = list_workflow_versions(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
"limit": 5,
},
)
print_json(versions_result)
print()
# Get workflow activities
print("Getting workflow activities...")
activities_result = get_workflow_activities(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(activities_result)
print()
# Add a new activity to the workflow
print("Adding a new approval activity to the workflow...")
add_activity_result = add_workflow_activity(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
"name": f"Demo Approval {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
"description": "A demo approval activity",
"activity_type": "approval",
},
)
print_json(add_activity_result)
print()
if "error" in add_activity_result:
print("Error adding activity. Skipping activity modification steps.")
else:
activity_id = add_activity_result["activity"]["sys_id"]
# Update the activity
print("Updating the activity...")
update_activity_result = update_workflow_activity(
auth_manager,
server_config,
{
"activity_id": activity_id,
"name": f"Updated Demo Approval {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
"description": "An updated demo approval activity",
},
)
print_json(update_activity_result)
print()
# Get the updated activities
print("Getting updated workflow activities...")
updated_activities_result = get_workflow_activities(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(updated_activities_result)
print()
# If there are multiple activities, reorder them
if updated_activities_result.get("count", 0) > 1:
print("Reordering workflow activities...")
activity_ids = [activity["sys_id"] for activity in updated_activities_result["activities"]]
# Reverse the order
activity_ids.reverse()
reorder_result = reorder_workflow_activities(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
"activity_ids": activity_ids,
},
)
print_json(reorder_result)
print()
# Get the reordered activities
print("Getting reordered workflow activities...")
reordered_activities_result = get_workflow_activities(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(reordered_activities_result)
print()
# Update the workflow
print("Updating the workflow...")
update_result = update_workflow(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
"description": f"Updated description {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
},
)
print_json(update_result)
print()
# Deactivate the workflow
print("Deactivating the workflow...")
deactivate_result = deactivate_workflow(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(deactivate_result)
print()
# Activate the workflow
print("Activating the workflow...")
activate_result = activate_workflow(
auth_manager,
server_config,
{
"workflow_id": workflow_id,
},
)
print_json(activate_result)
print()
print("Workflow management demo completed successfully!")
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/scripts/test_connection.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
ServiceNow Connection Test Script
This script tests the connection to a ServiceNow instance using the credentials
provided in the .env file. It supports all authentication methods:
- Basic authentication (username/password)
- OAuth authentication (client ID/client secret)
- API key authentication
Usage:
python scripts/test_connection.py
"""
import os
import sys
import requests
import base64
from pathlib import Path
from dotenv import load_dotenv
# Add the project root to the Python path
sys.path.insert(0, str(Path(__file__).parent.parent))
def get_oauth_token(instance_url, client_id, client_secret, username=None, password=None):
"""Get an OAuth token from ServiceNow."""
token_url = os.getenv("SERVICENOW_TOKEN_URL", f"{instance_url}/oauth_token.do")
# Create authorization header
auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
# Try different OAuth grant types
access_token = None
# 1. Try client credentials grant
try:
print("Attempting client_credentials grant...")
token_response = requests.post(
token_url,
headers={
"Authorization": f"Basic {auth_header}",
"Content-Type": "application/x-www-form-urlencoded"
},
data={
"grant_type": "client_credentials"
}
)
if token_response.status_code == 200:
token_data = token_response.json()
access_token = token_data.get("access_token")
print("✅ Successfully obtained OAuth token using client_credentials grant!")
return access_token
else:
print(f"❌ Failed with client_credentials grant: {token_response.status_code}")
print(f"Response: {token_response.text}")
except Exception as e:
print(f"❌ Error with client_credentials grant: {e}")
# 2. Try password grant if client credentials failed and we have username/password
if not access_token and username and password:
try:
print("Attempting password grant...")
token_response = requests.post(
token_url,
headers={
"Authorization": f"Basic {auth_header}",
"Content-Type": "application/x-www-form-urlencoded"
},
data={
"grant_type": "password",
"username": username,
"password": password
}
)
if token_response.status_code == 200:
token_data = token_response.json()
access_token = token_data.get("access_token")
print("✅ Successfully obtained OAuth token using password grant!")
return access_token
else:
print(f"❌ Failed with password grant: {token_response.status_code}")
print(f"Response: {token_response.text}")
except Exception as e:
print(f"❌ Error with password grant: {e}")
return None
def test_connection():
# Load environment variables
load_dotenv()
# Print all environment variables related to ServiceNow
print("Environment variables:")
for key, value in os.environ.items():
if "SERVICENOW" in key:
masked_value = value
if "PASSWORD" in key or "SECRET" in key:
masked_value = "*" * len(value)
print(f" {key}={masked_value}")
print()
# Get ServiceNow credentials
instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
auth_type = os.getenv("SERVICENOW_AUTH_TYPE", "basic")
# Check if instance URL is set
if not instance_url or instance_url == "https://your-instance.service-now.com":
print("Error: ServiceNow instance URL is not set or is using the default value.")
print("Please update the SERVICENOW_INSTANCE_URL in your .env file.")
sys.exit(1)
print(f"Testing connection to ServiceNow instance: {instance_url}")
print(f"Authentication type: {auth_type}")
# Construct API endpoint URL
api_url = f"{instance_url}/api/now/table/incident?sysparm_limit=1"
headers = {"Accept": "application/json"}
auth = None
# Set up authentication based on auth type
if auth_type == "basic":
username = os.getenv("SERVICENOW_USERNAME")
password = os.getenv("SERVICENOW_PASSWORD")
if not username or not password or username == "your-username" or password == "your-password":
print("Error: Username or password is not set or is using the default value.")
print("Please update the SERVICENOW_USERNAME and SERVICENOW_PASSWORD in your .env file.")
sys.exit(1)
auth = (username, password)
print("Using basic authentication (username/password)")
print(f"Username: {username}")
print(f"Password: {'*' * len(password)}")
elif auth_type == "oauth":
client_id = os.getenv("SERVICENOW_CLIENT_ID")
client_secret = os.getenv("SERVICENOW_CLIENT_SECRET")
username = os.getenv("SERVICENOW_USERNAME")
password = os.getenv("SERVICENOW_PASSWORD")
if not client_id or not client_secret or client_id == "your-client-id" or client_secret == "your-client-secret":
print("Error: Client ID or Client Secret is not set or is using the default value.")
print("Please update the SERVICENOW_CLIENT_ID and SERVICENOW_CLIENT_SECRET in your .env file.")
print("You can run scripts/setup_oauth.py to set up OAuth authentication.")
sys.exit(1)
print("Using OAuth authentication (client ID/client secret)")
access_token = get_oauth_token(instance_url, client_id, client_secret, username, password)
if not access_token:
print("Failed to obtain OAuth token. Please check your credentials.")
sys.exit(1)
headers["Authorization"] = f"Bearer {access_token}"
elif auth_type == "api_key":
api_key = os.getenv("SERVICENOW_API_KEY")
api_key_header = os.getenv("SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key")
if not api_key or api_key == "your-api-key":
print("Error: API key is not set or is using the default value.")
print("Please update the SERVICENOW_API_KEY in your .env file.")
print("You can run scripts/setup_api_key.py to set up API key authentication.")
sys.exit(1)
print(f"Using API key authentication (header: {api_key_header})")
headers[api_key_header] = api_key
else:
print(f"Error: Unsupported authentication type: {auth_type}")
print("Supported types: basic, oauth, api_key")
sys.exit(1)
try:
# Print request details
print("\nRequest details:")
print(f"URL: {api_url}")
print(f"Headers: {headers}")
if auth:
print(f"Auth: ({auth[0]}, {'*' * len(auth[1])})")
# Make a test request
if auth:
response = requests.get(api_url, auth=auth, headers=headers)
else:
response = requests.get(api_url, headers=headers)
# Print response details
print("\nResponse details:")
print(f"Status code: {response.status_code}")
print(f"Response headers: {dict(response.headers)}")
# Check response
if response.status_code == 200:
print("\n✅ Connection successful!")
data = response.json()
print(f"Retrieved {len(data.get('result', []))} incident(s)")
print("\nSample response:")
print(f"{response.text[:500]}...")
return True
else:
print(f"\n❌ Connection failed with status code: {response.status_code}")
print(f"Response: {response.text}")
return False
except requests.exceptions.RequestException as e:
print(f"\n❌ Connection error: {e}")
return False
if __name__ == "__main__":
test_connection()
```
--------------------------------------------------------------------------------
/docs/workflow_management.md:
--------------------------------------------------------------------------------
```markdown
# Workflow Management in ServiceNow MCP
This document provides detailed information about the workflow management tools available in the ServiceNow MCP server.
## Overview
ServiceNow workflows are a powerful automation feature that allows you to define and automate business processes. The workflow management tools in the ServiceNow MCP server enable you to view, create, and modify workflows in your ServiceNow instance.
## Available Tools
### Viewing Workflows
1. **list_workflows** - List workflows from ServiceNow
- Parameters:
- `limit` (optional): Maximum number of records to return (default: 10)
- `offset` (optional): Offset to start from (default: 0)
- `active` (optional): Filter by active status (true/false)
- `name` (optional): Filter by name (contains)
- `query` (optional): Additional query string
2. **get_workflow_details** - Get detailed information about a specific workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
3. **list_workflow_versions** - List all versions of a specific workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
- `limit` (optional): Maximum number of records to return (default: 10)
- `offset` (optional): Offset to start from (default: 0)
4. **get_workflow_activities** - Get all activities in a workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
- `version` (optional): Specific version to get activities for (if not provided, the latest published version will be used)
### Modifying Workflows
5. **create_workflow** - Create a new workflow in ServiceNow
- Parameters:
- `name` (required): Name of the workflow
- `description` (optional): Description of the workflow
- `table` (optional): Table the workflow applies to
- `active` (optional): Whether the workflow is active (default: true)
- `attributes` (optional): Additional attributes for the workflow
6. **update_workflow** - Update an existing workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
- `name` (optional): Name of the workflow
- `description` (optional): Description of the workflow
- `table` (optional): Table the workflow applies to
- `active` (optional): Whether the workflow is active
- `attributes` (optional): Additional attributes for the workflow
7. **activate_workflow** - Activate a workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
8. **deactivate_workflow** - Deactivate a workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
### Managing Workflow Activities
9. **add_workflow_activity** - Add a new activity to a workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
- `name` (required): Name of the activity
- `description` (optional): Description of the activity
- `activity_type` (required): Type of activity (e.g., 'approval', 'task', 'notification')
- `attributes` (optional): Additional attributes for the activity
- `position` (optional): Position in the workflow (if not provided, the activity will be added at the end)
10. **update_workflow_activity** - Update an existing activity in a workflow
- Parameters:
- `activity_id` (required): Activity ID or sys_id
- `name` (optional): Name of the activity
- `description` (optional): Description of the activity
- `attributes` (optional): Additional attributes for the activity
11. **delete_workflow_activity** - Delete an activity from a workflow
- Parameters:
- `activity_id` (required): Activity ID or sys_id
12. **reorder_workflow_activities** - Change the order of activities in a workflow
- Parameters:
- `workflow_id` (required): Workflow ID or sys_id
- `activity_ids` (required): List of activity IDs in the desired order
## Usage Examples
### Viewing Workflows
#### List all active workflows
```python
result = list_workflows({
"active": True,
"limit": 20
})
```
#### Get details about a specific workflow
```python
result = get_workflow_details({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
})
```
#### List all versions of a workflow
```python
result = list_workflow_versions({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
})
```
#### Get all activities in a workflow
```python
result = get_workflow_activities({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
})
```
### Modifying Workflows
#### Create a new workflow
```python
result = create_workflow({
"name": "Software License Request",
"description": "Workflow for handling software license requests",
"table": "sc_request"
})
```
#### Update an existing workflow
```python
result = update_workflow({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
"description": "Updated workflow description",
"active": True
})
```
#### Activate a workflow
```python
result = activate_workflow({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
})
```
#### Deactivate a workflow
```python
result = deactivate_workflow({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
})
```
### Managing Workflow Activities
#### Add a new activity to a workflow
```python
result = add_workflow_activity({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
"name": "Manager Approval",
"description": "Approval step for the manager",
"activity_type": "approval"
})
```
#### Update an existing activity
```python
result = update_workflow_activity({
"activity_id": "3cda7cda87a9c150e0b0df23cebb3591",
"name": "Updated Activity Name",
"description": "Updated activity description"
})
```
#### Delete an activity
```python
result = delete_workflow_activity({
"activity_id": "3cda7cda87a9c150e0b0df23cebb3591"
})
```
#### Reorder activities in a workflow
```python
result = reorder_workflow_activities({
"workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
"activity_ids": [
"3cda7cda87a9c150e0b0df23cebb3591",
"4cda7cda87a9c150e0b0df23cebb3592",
"5cda7cda87a9c150e0b0df23cebb3593"
]
})
```
## Common Activity Types
ServiceNow provides several activity types that can be used when adding activities to a workflow:
1. **approval** - An approval activity that requires user action
2. **task** - A task that needs to be completed
3. **notification** - Sends a notification to users
4. **timer** - Waits for a specified amount of time
5. **condition** - Evaluates a condition and branches the workflow
6. **script** - Executes a script
7. **wait_for_condition** - Waits until a condition is met
8. **end** - Ends the workflow
## Best Practices
1. **Version Control**: Always create a new version of a workflow before making significant changes.
2. **Testing**: Test workflows in a non-production environment before deploying to production.
3. **Documentation**: Document the purpose and behavior of each workflow and activity.
4. **Error Handling**: Include error handling in your workflows to handle unexpected situations.
5. **Notifications**: Use notification activities to keep stakeholders informed about the workflow progress.
## Troubleshooting
### Common Issues
1. **Error: "No published versions found for this workflow"**
- This error occurs when trying to get activities for a workflow that has no published versions.
- Solution: Publish a version of the workflow before trying to get its activities.
2. **Error: "Activity type is required"**
- This error occurs when trying to add an activity without specifying its type.
- Solution: Provide a valid activity type when adding an activity.
3. **Error: "Cannot modify a published workflow version"**
- This error occurs when trying to modify a published workflow version.
- Solution: Create a new draft version of the workflow before making changes.
4. **Error: "Workflow ID is required"**
- This error occurs when not providing a workflow ID for operations that require it.
- Solution: Make sure to include the workflow ID in your request.
## Additional Resources
- [ServiceNow Workflow Documentation](https://docs.servicenow.com/bundle/tokyo-platform-administration/page/administer/workflow-administration/concept/c_WorkflowAdministration.html)
- [ServiceNow Workflow API Reference](https://developer.servicenow.com/dev.do#!/reference/api/tokyo/rest/c_WorkflowAPI)
```
--------------------------------------------------------------------------------
/docs/catalog_optimization_plan.md:
--------------------------------------------------------------------------------
```markdown
# ServiceNow Catalog Optimization Plan
This document outlines the plan for implementing catalog optimization features in the ServiceNow MCP integration.
## Overview
The catalog optimization features will help users analyze, manage, and improve their existing ServiceNow service catalogs. These features will provide insights into catalog usage, performance, and structure, and offer recommendations for optimization.
## Optimization Tools
### 1. Catalog Analytics Tools
#### 1.1 Get Catalog Usage Statistics
**Parameters:**
```python
class CatalogUsageStatsParams(BaseModel):
"""Parameters for getting catalog usage statistics."""
time_period: str = Field("last_30_days", description="Time period for statistics (last_7_days, last_30_days, last_90_days, last_year)")
category_id: Optional[str] = Field(None, description="Filter by category ID")
include_inactive: bool = Field(False, description="Whether to include inactive items")
```
**Returns:**
- Most ordered items
- Least ordered items
- Average processing time
- Abandonment rate (items added to cart but not ordered)
- User satisfaction ratings
**Implementation:**
- Query the ServiceNow API for order statistics
- Aggregate data by item and category
- Calculate metrics like order volume, processing time, and abandonment rate
#### 1.2 Get Item Performance Metrics
**Parameters:**
```python
class ItemPerformanceParams(BaseModel):
"""Parameters for getting performance metrics for a catalog item."""
item_id: str = Field(..., description="Catalog item ID")
time_period: str = Field("last_30_days", description="Time period for metrics")
```
**Returns:**
- Order volume over time
- Average fulfillment time
- Approval rates
- Rejection reasons
- User ratings and feedback
**Implementation:**
- Query the ServiceNow API for item-specific metrics
- Calculate performance indicators
- Identify trends over the specified time period
### 2. Catalog Management Tools
#### 2.1 Update Catalog Item
**Parameters:**
```python
class UpdateCatalogItemParams(BaseModel):
"""Parameters for updating a catalog item."""
item_id: str = Field(..., description="Catalog item ID to update")
name: Optional[str] = Field(None, description="New name for the item")
short_description: Optional[str] = Field(None, description="New short description")
description: Optional[str] = Field(None, description="New detailed description")
category: Optional[str] = Field(None, description="New category ID")
price: Optional[str] = Field(None, description="New price")
active: Optional[bool] = Field(None, description="Whether the item is active")
order: Optional[int] = Field(None, description="Display order in the category")
```
**Implementation:**
- Build a PATCH request to update the catalog item
- Only include fields that are provided in the parameters
- Return the updated item details
#### 2.2 Update Catalog Category
**Parameters:**
```python
class UpdateCatalogCategoryParams(BaseModel):
"""Parameters for updating a catalog category."""
category_id: str = Field(..., description="Category ID to update")
title: Optional[str] = Field(None, description="New title for the category")
description: Optional[str] = Field(None, description="New description")
parent: Optional[str] = Field(None, description="New parent category ID")
active: Optional[bool] = Field(None, description="Whether the category is active")
order: Optional[int] = Field(None, description="Display order")
```
**Implementation:**
- Build a PATCH request to update the catalog category
- Only include fields that are provided in the parameters
- Return the updated category details
#### 2.3 Update Item Variable
**Parameters:**
```python
class UpdateItemVariableParams(BaseModel):
"""Parameters for updating a catalog item variable."""
variable_id: str = Field(..., description="Variable ID to update")
label: Optional[str] = Field(None, description="New label for the variable")
help_text: Optional[str] = Field(None, description="New help text")
default_value: Optional[str] = Field(None, description="New default value")
mandatory: Optional[bool] = Field(None, description="Whether the variable is mandatory")
order: Optional[int] = Field(None, description="Display order")
```
**Implementation:**
- Build a PATCH request to update the catalog item variable
- Only include fields that are provided in the parameters
- Return the updated variable details
### 3. Catalog Optimization Tools
#### 3.1 Get Optimization Recommendations
**Parameters:**
```python
class OptimizationRecommendationsParams(BaseModel):
"""Parameters for getting catalog optimization recommendations."""
category_id: Optional[str] = Field(None, description="Filter by category ID")
recommendation_types: List[str] = Field(
["inactive_items", "low_usage", "high_abandonment", "slow_fulfillment", "description_quality"],
description="Types of recommendations to include"
)
```
**Returns:**
- Inactive items that could be retired
- Items with low usage that might need promotion
- Items with high abandonment rates that might need simplification
- Items with slow fulfillment that need process improvements
- Items with poor description quality
**Implementation:**
- Query the ServiceNow API for various metrics
- Apply analysis algorithms to identify optimization opportunities
- Generate recommendations based on the analysis
#### 3.2 Get Catalog Structure Analysis
**Parameters:**
```python
class CatalogStructureAnalysisParams(BaseModel):
"""Parameters for analyzing catalog structure."""
include_inactive: bool = Field(False, description="Whether to include inactive categories and items")
```
**Returns:**
- Categories with too many or too few items
- Deeply nested categories that might be hard to navigate
- Inconsistent naming patterns
- Duplicate or similar items across categories
**Implementation:**
- Query the ServiceNow API for the catalog structure
- Analyze the structure for usability issues
- Generate recommendations for improving the structure
## Implementation Plan
### Phase 1: Analytics Tools
1. Implement `get_catalog_usage_stats`
2. Implement `get_item_performance`
3. Create tests for analytics tools
### Phase 2: Management Tools
1. Implement `update_catalog_item`
2. Implement `update_catalog_category`
3. Implement `update_item_variable`
4. Create tests for management tools
### Phase 3: Optimization Tools
1. Implement `get_optimization_recommendations`
2. Implement `get_catalog_structure_analysis`
3. Create tests for optimization tools
### Phase 4: Integration
1. Register all tools with the MCP server
2. Create example scripts for optimization workflows
3. Update documentation
## Example Usage
### Example 1: Analyzing Catalog Usage
```python
# Get catalog usage statistics
params = CatalogUsageStatsParams(time_period="last_90_days")
result = get_catalog_usage_stats(config, auth_manager, params)
# Print the most ordered items
print("Most ordered items:")
for item in result["most_ordered_items"]:
print(f"- {item['name']}: {item['order_count']} orders")
# Print items with high abandonment rates
print("\nItems with high abandonment rates:")
for item in result["high_abandonment_items"]:
print(f"- {item['name']}: {item['abandonment_rate']}% abandonment rate")
```
### Example 2: Getting Optimization Recommendations
```python
# Get optimization recommendations
params = OptimizationRecommendationsParams(
recommendation_types=["inactive_items", "low_usage", "description_quality"]
)
result = get_optimization_recommendations(config, auth_manager, params)
# Print the recommendations
for rec in result["recommendations"]:
print(f"\n{rec['title']}")
print(f"Impact: {rec['impact']}, Effort: {rec['effort']}")
print(f"{rec['description']}")
print(f"Recommended Action: {rec['action']}")
print(f"Affected Items: {len(rec['items'])}")
for item in rec['items'][:3]:
print(f"- {item['name']}: {item['short_description']}")
```
### Example 3: Updating a Catalog Item
```python
# Update a catalog item
params = UpdateCatalogItemParams(
item_id="sys_id_of_item",
short_description="Updated description that is more clear and informative",
price="99.99"
)
result = update_catalog_item(config, auth_manager, params)
if result["success"]:
print(f"Successfully updated item: {result['data']['name']}")
else:
print(f"Error: {result['message']}")
```
## Considerations
1. **Data Access**: These tools require access to ServiceNow reporting and analytics data, which might require additional permissions.
2. **Performance**: Some of these analyses could be resource-intensive, especially for large catalogs.
3. **Custom Metrics**: ServiceNow instances often have custom metrics and KPIs for catalog performance, which would need to be considered.
4. **Change Management**: Any changes to the catalog should follow proper change management processes.
5. **User Feedback**: Incorporating user feedback data would make the optimization recommendations more valuable.
```
--------------------------------------------------------------------------------
/tests/test_script_include_resources.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the script include resources.
This module contains tests for the script include resources in the ServiceNow MCP server.
"""
import json
import unittest
import requests
from unittest.mock import MagicMock, patch
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.resources.script_includes import ScriptIncludeListParams, ScriptIncludeResource
from servicenow_mcp.utils.config import ServerConfig, AuthConfig, AuthType, BasicAuthConfig
class TestScriptIncludeResource(unittest.IsolatedAsyncioTestCase):
"""Tests for the script include resource."""
def setUp(self):
"""Set up test fixtures."""
auth_config = AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(
username="test_user",
password="test_password"
)
)
self.server_config = ServerConfig(
instance_url="https://test.service-now.com",
auth=auth_config,
)
self.auth_manager = MagicMock(spec=AuthManager)
self.auth_manager.get_headers.return_value = {"Authorization": "Bearer test"}
self.script_include_resource = ScriptIncludeResource(self.server_config, self.auth_manager)
@patch("servicenow_mcp.resources.script_includes.requests.get")
async def test_list_script_includes(self, mock_get):
"""Test listing script includes."""
# Mock response
mock_response = MagicMock()
mock_response.text = json.dumps({
"result": [
{
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": "true",
"active": "true",
"access": "public",
"sys_created_on": "2023-01-01 00:00:00",
"sys_updated_on": "2023-01-02 00:00:00",
"sys_created_by": {"display_value": "admin"},
"sys_updated_by": {"display_value": "admin"}
}
]
})
mock_response.status_code = 200
mock_get.return_value = mock_response
# Call the method
params = ScriptIncludeListParams(
limit=10,
offset=0,
active=True,
client_callable=True,
query="Test"
)
result = await self.script_include_resource.list_script_includes(params)
result_json = json.loads(result)
# Verify the result
self.assertIn("result", result_json)
self.assertEqual(1, len(result_json["result"]))
self.assertEqual("123", result_json["result"][0]["sys_id"])
self.assertEqual("TestScriptInclude", result_json["result"][0]["name"])
self.assertEqual("true", result_json["result"][0]["client_callable"])
self.assertEqual("true", result_json["result"][0]["active"])
# Verify the request
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
self.assertEqual(10, kwargs["params"]["sysparm_limit"])
self.assertEqual(0, kwargs["params"]["sysparm_offset"])
self.assertEqual("active=true^client_callable=true^nameLIKETest", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.resources.script_includes.requests.get")
async def test_get_script_include(self, mock_get):
"""Test getting a script include."""
# Mock response
mock_response = MagicMock()
mock_response.text = json.dumps({
"result": {
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": "true",
"active": "true",
"access": "public",
"sys_created_on": "2023-01-01 00:00:00",
"sys_updated_on": "2023-01-02 00:00:00",
"sys_created_by": {"display_value": "admin"},
"sys_updated_by": {"display_value": "admin"}
}
})
mock_response.status_code = 200
mock_get.return_value = mock_response
# Call the method
result = await self.script_include_resource.get_script_include("123")
result_json = json.loads(result)
# Verify the result
self.assertIn("result", result_json)
self.assertEqual("123", result_json["result"]["sys_id"])
self.assertEqual("TestScriptInclude", result_json["result"]["name"])
self.assertEqual("true", result_json["result"]["client_callable"])
self.assertEqual("true", result_json["result"]["active"])
# Verify the request
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
self.assertEqual("name=123", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.resources.script_includes.requests.get")
async def test_get_script_include_by_sys_id(self, mock_get):
"""Test getting a script include by sys_id."""
# Mock response
mock_response = MagicMock()
mock_response.text = json.dumps({
"result": {
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": "true",
"active": "true",
"access": "public",
"sys_created_on": "2023-01-01 00:00:00",
"sys_updated_on": "2023-01-02 00:00:00",
"sys_created_by": {"display_value": "admin"},
"sys_updated_by": {"display_value": "admin"}
}
})
mock_response.status_code = 200
mock_get.return_value = mock_response
# Call the method
result = await self.script_include_resource.get_script_include("sys_id:123")
result_json = json.loads(result)
# Verify the result
self.assertIn("result", result_json)
self.assertEqual("123", result_json["result"]["sys_id"])
self.assertEqual("TestScriptInclude", result_json["result"]["name"])
# Verify the request
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include/123", args[0])
self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
@patch("servicenow_mcp.resources.script_includes.requests.get")
async def test_list_script_includes_error(self, mock_get):
"""Test listing script includes with an error."""
# Mock response
mock_get.side_effect = requests.RequestException("Test error")
# Call the method
params = ScriptIncludeListParams()
result = await self.script_include_resource.list_script_includes(params)
result_json = json.loads(result)
# Verify the result
self.assertIn("error", result_json)
self.assertIn("Error listing script includes", result_json["error"])
@patch("servicenow_mcp.resources.script_includes.requests.get")
async def test_get_script_include_error(self, mock_get):
"""Test getting a script include with an error."""
# Mock response
mock_get.side_effect = requests.RequestException("Test error")
# Call the method
result = await self.script_include_resource.get_script_include("123")
result_json = json.loads(result)
# Verify the result
self.assertIn("error", result_json)
self.assertIn("Error getting script include", result_json["error"])
class TestScriptIncludeListParams(unittest.TestCase):
"""Tests for the script include list parameters."""
def test_script_include_list_params(self):
"""Test script include list parameters."""
params = ScriptIncludeListParams(
limit=20,
offset=10,
active=True,
client_callable=False,
query="Test"
)
self.assertEqual(20, params.limit)
self.assertEqual(10, params.offset)
self.assertTrue(params.active)
self.assertFalse(params.client_callable)
self.assertEqual("Test", params.query)
def test_script_include_list_params_defaults(self):
"""Test script include list parameters defaults."""
params = ScriptIncludeListParams()
self.assertEqual(10, params.limit)
self.assertEqual(0, params.offset)
self.assertIsNone(params.active)
self.assertIsNone(params.client_callable)
self.assertIsNone(params.query)
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/cli.py:
--------------------------------------------------------------------------------
```python
"""
Command-line interface for the ServiceNow MCP server.
"""
import argparse
import logging
import os
import sys
import anyio
from dotenv import load_dotenv
from mcp.server.stdio import stdio_server
from servicenow_mcp.server import ServiceNowMCP
from servicenow_mcp.utils.config import (
ApiKeyConfig,
AuthConfig,
AuthType,
BasicAuthConfig,
OAuthConfig,
ServerConfig,
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def parse_args():
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(description="ServiceNow MCP Server")
# Server configuration
parser.add_argument(
"--instance-url",
help="ServiceNow instance URL (e.g., https://instance.service-now.com)",
default=os.environ.get("SERVICENOW_INSTANCE_URL"),
)
parser.add_argument(
"--debug",
action="store_true",
help="Enable debug mode",
default=os.environ.get("SERVICENOW_DEBUG", "false").lower() == "true",
)
parser.add_argument(
"--timeout",
type=int,
help="Request timeout in seconds",
default=int(os.environ.get("SERVICENOW_TIMEOUT", "30")),
)
# Authentication
auth_group = parser.add_argument_group("Authentication")
auth_group.add_argument(
"--auth-type",
choices=["basic", "oauth", "api_key"],
help="Authentication type",
default=os.environ.get("SERVICENOW_AUTH_TYPE", "basic"),
)
# Basic auth
basic_group = parser.add_argument_group("Basic Authentication")
basic_group.add_argument(
"--username",
help="ServiceNow username",
default=os.environ.get("SERVICENOW_USERNAME"),
)
basic_group.add_argument(
"--password",
help="ServiceNow password",
default=os.environ.get("SERVICENOW_PASSWORD"),
)
# OAuth
oauth_group = parser.add_argument_group("OAuth Authentication")
oauth_group.add_argument(
"--client-id",
help="OAuth client ID",
default=os.environ.get("SERVICENOW_CLIENT_ID"),
)
oauth_group.add_argument(
"--client-secret",
help="OAuth client secret",
default=os.environ.get("SERVICENOW_CLIENT_SECRET"),
)
oauth_group.add_argument(
"--token-url",
help="OAuth token URL",
default=os.environ.get("SERVICENOW_TOKEN_URL"),
)
# API Key
api_key_group = parser.add_argument_group("API Key Authentication")
api_key_group.add_argument(
"--api-key",
help="ServiceNow API key",
default=os.environ.get("SERVICENOW_API_KEY"),
)
api_key_group.add_argument(
"--api-key-header",
help="API key header name",
default=os.environ.get("SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key"),
)
# Script execution API resource path
script_execution_group = parser.add_argument_group("Script Execution API")
script_execution_group.add_argument(
"--script-execution-api-resource-path",
help="Script execution API resource path",
default=os.environ.get("SCRIPT_EXECUTION_API_RESOURCE_PATH"),
)
return parser.parse_args()
def create_config(args) -> ServerConfig:
"""
Create server configuration from command-line arguments.
Args:
args: Command-line arguments.
Returns:
ServerConfig: Server configuration.
Raises:
ValueError: If required configuration is missing.
"""
# NOTE: This assumes the ServerConfig model takes instance_url, auth, debug, timeout etc.
# The ServiceNowMCP class now expects a ServerConfig object matching this.
# Instance URL validation
instance_url = args.instance_url
if not instance_url:
# Attempt to load from .env if not provided via args/env vars directly in parse_args
instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
if not instance_url:
raise ValueError(
"ServiceNow instance URL is required (--instance-url or SERVICENOW_INSTANCE_URL env var)"
)
# Create authentication configuration based on args
auth_type = AuthType(args.auth_type.lower())
# This will hold the final AuthConfig instance for ServerConfig
final_auth_config: AuthConfig
if auth_type == AuthType.BASIC:
username = args.username or os.getenv("SERVICENOW_USERNAME")
password = args.password or os.getenv("SERVICENOW_PASSWORD") # Get password from arg or env
if not username or not password:
raise ValueError(
"Username and password are required for basic authentication (--username/SERVICENOW_USERNAME, --password/SERVICENOW_PASSWORD)"
)
# Create the specific config (without instance_url)
basic_cfg = BasicAuthConfig(
username=username,
password=password,
)
# Create the main AuthConfig wrapper
final_auth_config = AuthConfig(type=auth_type, basic=basic_cfg)
elif auth_type == AuthType.OAUTH:
# Simplified - assuming password grant for now based on previous args
client_id = args.client_id or os.getenv("SERVICENOW_CLIENT_ID")
client_secret = args.client_secret or os.getenv("SERVICENOW_CLIENT_SECRET")
username = args.username or os.getenv("SERVICENOW_USERNAME") # Needed for password grant
password = args.password or os.getenv("SERVICENOW_PASSWORD") # Needed for password grant
token_url = args.token_url or os.getenv("SERVICENOW_TOKEN_URL")
if not client_id or not client_secret or not username or not password:
raise ValueError(
"Client ID, client secret, username, and password are required for OAuth password grant"
" (--client-id/SERVICENOW_CLIENT_ID, etc.)"
)
if not token_url:
# Attempt to construct default if not provided
token_url = f"{instance_url}/oauth_token.do"
logger.warning(f"OAuth token URL not provided, defaulting to: {token_url}")
# Create the specific config (without instance_url)
oauth_cfg = OAuthConfig(
client_id=client_id,
client_secret=client_secret,
username=username,
password=password,
token_url=token_url,
)
# Create the main AuthConfig wrapper
final_auth_config = AuthConfig(type=auth_type, oauth=oauth_cfg)
elif auth_type == AuthType.API_KEY:
api_key = args.api_key or os.getenv("SERVICENOW_API_KEY")
api_key_header = args.api_key_header or os.getenv(
"SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key"
)
if not api_key:
raise ValueError(
"API key is required for API key authentication (--api-key or SERVICENOW_API_KEY)"
)
# Create the specific config (without instance_url)
api_key_cfg = ApiKeyConfig(
api_key=api_key,
header_name=api_key_header,
)
# Create the main AuthConfig wrapper
final_auth_config = AuthConfig(type=auth_type, api_key=api_key_cfg)
else:
# Should not happen if choices are enforced by argparse
raise ValueError(f"Unsupported authentication type: {args.auth_type}")
# Script execution path
script_execution_api_resource_path = args.script_execution_api_resource_path or os.getenv(
"SCRIPT_EXECUTION_API_RESOURCE_PATH"
)
if not script_execution_api_resource_path:
logger.warning(
"Script execution API resource path not set (--script-execution-api-resource-path or SCRIPT_EXECUTION_API_RESOURCE_PATH). ExecuteScriptInclude tool may fail."
)
# Create the final ServerConfig
# Ensure ServerConfig model expects 'auth' as a nested object
return ServerConfig(
instance_url=instance_url, # Add instance_url directly here
auth=final_auth_config, # Pass the correctly structured AuthConfig instance
# Include other server config fields if they exist on ServerConfig model
debug=args.debug,
timeout=args.timeout,
script_execution_api_resource_path=script_execution_api_resource_path,
)
async def arun_server(server_instance):
"""Runs the given MCP server instance using stdio transport."""
logger.info("Starting server with stdio transport...")
async with stdio_server() as streams:
# Get initialization options from the low-level server
init_options = server_instance.create_initialization_options()
await server_instance.run(streams[0], streams[1], init_options)
logger.info("Stdio server finished.")
def main():
"""Main entry point for the CLI."""
# Load environment variables from .env file
load_dotenv()
try:
# Parse command-line arguments
args = parse_args()
# Configure logging level based on debug flag
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
logger.info("Debug logging enabled.")
else:
logging.getLogger().setLevel(logging.INFO)
# Create server configuration
config = create_config(args)
# Log the instance URL being used (mask sensitive parts of config if needed)
logger.info(f"Initializing ServiceNow MCP server for instance: {config.instance_url}")
# Create server controller instance
mcp_controller = ServiceNowMCP(config)
# Get the low-level server instance to run
server_to_run = mcp_controller.start()
# Run the server using anyio and the stdio transport
anyio.run(arun_server, server_to_run)
except ValueError as e:
logger.error(f"Configuration or runtime error: {e}")
sys.exit(1)
except Exception as e:
logger.exception(f"Unexpected error starting or running server: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/docs/user_management.md:
--------------------------------------------------------------------------------
```markdown
# User Management in ServiceNow MCP
This document provides detailed information about the User Management tools available in the ServiceNow MCP server.
## Overview
The User Management tools allow you to create, update, and manage users and groups in ServiceNow. These tools are essential for setting up test environments, creating users with specific roles, and organizing users into assignment groups.
## Available Tools
### User Management
1. **create_user** - Create a new user in ServiceNow
2. **update_user** - Update an existing user in ServiceNow
3. **get_user** - Get a specific user by ID, username, or email
4. **list_users** - List users with filtering options
### Group Management
5. **create_group** - Create a new group in ServiceNow
6. **update_group** - Update an existing group in ServiceNow
7. **add_group_members** - Add members to a group in ServiceNow
8. **remove_group_members** - Remove members from a group in ServiceNow
9. **list_groups** - List groups with filtering options
## Tool Details
### create_user
Creates a new user in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| user_name | string | Yes | Username for the user |
| first_name | string | Yes | First name of the user |
| last_name | string | Yes | Last name of the user |
| email | string | Yes | Email address of the user |
| title | string | No | Job title of the user |
| department | string | No | Department the user belongs to |
| manager | string | No | Manager of the user (sys_id or username) |
| roles | array | No | Roles to assign to the user |
| phone | string | No | Phone number of the user |
| mobile_phone | string | No | Mobile phone number of the user |
| location | string | No | Location of the user |
| password | string | No | Password for the user account |
| active | boolean | No | Whether the user account is active (default: true) |
#### Example
```python
# Create a new user in the Radiology department
result = create_user({
"user_name": "alice.radiology",
"first_name": "Alice",
"last_name": "Radiology",
"email": "[email protected]",
"title": "Doctor",
"department": "Radiology",
"roles": ["user"]
})
```
### update_user
Updates an existing user in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| user_id | string | Yes | User ID or sys_id to update |
| user_name | string | No | Username for the user |
| first_name | string | No | First name of the user |
| last_name | string | No | Last name of the user |
| email | string | No | Email address of the user |
| title | string | No | Job title of the user |
| department | string | No | Department the user belongs to |
| manager | string | No | Manager of the user (sys_id or username) |
| roles | array | No | Roles to assign to the user |
| phone | string | No | Phone number of the user |
| mobile_phone | string | No | Mobile phone number of the user |
| location | string | No | Location of the user |
| password | string | No | Password for the user account |
| active | boolean | No | Whether the user account is active |
#### Example
```python
# Update a user to set their manager
result = update_user({
"user_id": "user123",
"manager": "user456",
"title": "Senior Doctor"
})
```
### get_user
Gets a specific user from ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| user_id | string | No | User ID or sys_id |
| user_name | string | No | Username of the user |
| email | string | No | Email address of the user |
**Note**: At least one of the parameters must be provided.
#### Example
```python
# Get a user by username
result = get_user({
"user_name": "alice.radiology"
})
```
### list_users
Lists users from ServiceNow with filtering options.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| limit | integer | No | Maximum number of users to return (default: 10) |
| offset | integer | No | Offset for pagination (default: 0) |
| active | boolean | No | Filter by active status |
| department | string | No | Filter by department |
| query | string | No | Search query for users |
#### Example
```python
# List users in the Radiology department
result = list_users({
"department": "Radiology",
"active": true,
"limit": 20
})
```
### create_group
Creates a new group in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| name | string | Yes | Name of the group |
| description | string | No | Description of the group |
| manager | string | No | Manager of the group (sys_id or username) |
| parent | string | No | Parent group (sys_id or name) |
| type | string | No | Type of the group |
| email | string | No | Email address for the group |
| members | array | No | List of user sys_ids or usernames to add as members |
| active | boolean | No | Whether the group is active (default: true) |
#### Example
```python
# Create a new group for Biomedical Engineering
result = create_group({
"name": "Biomedical Engineering",
"description": "Group for biomedical engineering staff",
"manager": "user456",
"members": ["admin", "alice.radiology"]
})
```
### update_group
Updates an existing group in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| group_id | string | Yes | Group ID or sys_id to update |
| name | string | No | Name of the group |
| description | string | No | Description of the group |
| manager | string | No | Manager of the group (sys_id or username) |
| parent | string | No | Parent group (sys_id or name) |
| type | string | No | Type of the group |
| email | string | No | Email address for the group |
| active | boolean | No | Whether the group is active |
#### Example
```python
# Update a group to change its manager
result = update_group({
"group_id": "group123",
"description": "Updated description for biomedical engineering group",
"manager": "user789"
})
```
### add_group_members
Adds members to a group in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| group_id | string | Yes | Group ID or sys_id |
| members | array | Yes | List of user sys_ids or usernames to add as members |
#### Example
```python
# Add members to the Biomedical Engineering group
result = add_group_members({
"group_id": "group123",
"members": ["bob.chiefradiology", "admin"]
})
```
### remove_group_members
Removes members from a group in ServiceNow.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| group_id | string | Yes | Group ID or sys_id |
| members | array | Yes | List of user sys_ids or usernames to remove as members |
#### Example
```python
# Remove a member from the Biomedical Engineering group
result = remove_group_members({
"group_id": "group123",
"members": ["alice.radiology"]
})
```
### list_groups
Lists groups from ServiceNow with filtering options.
#### Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| limit | integer | No | Maximum number of groups to return (default: 10) |
| offset | integer | No | Offset for pagination (default: 0) |
| active | boolean | No | Filter by active status |
| type | string | No | Filter by group type |
| query | string | No | Case-insensitive search term that matches against group name or description fields. Uses ServiceNow's LIKE operator for partial matching. |
#### Example
```python
# List active IT-type groups
result = list_groups({
"active": true,
"type": "it",
"query": "support",
"limit": 20
})
```
## Common Scenarios
### Creating Test Users and Groups for Approval Workflows
To set up test users and groups for an approval workflow:
1. Create department head user:
```python
bob_result = create_user({
"user_name": "bob.chiefradiology",
"first_name": "Bob",
"last_name": "ChiefRadiology",
"email": "[email protected]",
"title": "Chief of Radiology",
"department": "Radiology",
"roles": ["itil", "admin"] # assign ITIL role for approvals
})
```
2. Create staff user with department head as manager:
```python
alice_result = create_user({
"user_name": "alice.radiology",
"first_name": "Alice",
"last_name": "Radiology",
"email": "[email protected]",
"title": "Doctor",
"department": "Radiology",
"manager": bob_result.user_id # Set Bob as Alice's manager
})
```
3. Create assignment group for fulfillment:
```python
group_result = create_group({
"name": "Biomedical Engineering",
"description": "Group for biomedical engineering staff",
"members": ["admin"] # Add administrator as a member
})
```
### Finding Users in a Department
To find all users in a specific department:
```python
users = list_users({
"department": "Radiology",
"limit": 50
})
```
### Setting up Role-Based Access Control
To assign specific roles to users:
```python
# Assign ITIL role to a user so they can approve changes
update_user({
"user_id": "user123",
"roles": ["itil"]
})
```
## Troubleshooting
### Common Errors
1. **User already exists**
- This error occurs when trying to create a user with a username that already exists
- Solution: Use a different username or update the existing user instead
2. **User not found**
- This error occurs when trying to update, get, or add a user that doesn't exist
- Solution: Verify the user ID, username, or email is correct
3. **Role not found**
- This error occurs when trying to assign a role that doesn't exist
- Solution: Check the role name and make sure it exists in the ServiceNow instance
4. **Group not found**
- This error occurs when trying to update or add members to a group that doesn't exist
- Solution: Verify the group ID is correct
## Best Practices
1. **Use meaningful usernames**: Create usernames that reflect the user's identity, such as "firstname.lastname"
2. **Set up proper role assignments**: Only assign the necessary roles to users to maintain security best practices
3. **Organize users into appropriate groups**: Use groups to organize users based on departments, functions, or teams
4. **Manage group memberships carefully**: Add or remove users from groups to ensure proper assignment and notification routing
5. **Set managers for hierarchical approval flows**: When creating users that will be part of approval workflows, make sure to set the manager field appropriately
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/catalog_variables.py:
--------------------------------------------------------------------------------
```python
"""
Catalog Item Variables tools for the ServiceNow MCP server.
This module provides tools for managing variables (form fields) in ServiceNow catalog items.
"""
import logging
from typing import Any, Dict, List, 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 CreateCatalogItemVariableParams(BaseModel):
"""Parameters for creating a catalog item variable."""
catalog_item_id: str = Field(..., description="The sys_id of the catalog item")
name: str = Field(..., description="The name of the variable (internal name)")
type: str = Field(..., description="The type of variable (e.g., string, integer, boolean, reference)")
label: str = Field(..., description="The display label for the variable")
mandatory: bool = Field(False, description="Whether the variable is required")
help_text: Optional[str] = Field(None, description="Help text to display with the variable")
default_value: Optional[str] = Field(None, description="Default value for the variable")
description: Optional[str] = Field(None, description="Description of the variable")
order: Optional[int] = Field(None, description="Display order of the variable")
reference_table: Optional[str] = Field(None, description="For reference fields, the table to reference")
reference_qualifier: Optional[str] = Field(None, description="For reference fields, the query to filter reference options")
max_length: Optional[int] = Field(None, description="Maximum length for string fields")
min: Optional[int] = Field(None, description="Minimum value for numeric fields")
max: Optional[int] = Field(None, description="Maximum value for numeric fields")
class CatalogItemVariableResponse(BaseModel):
"""Response from catalog item variable operations."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
variable_id: Optional[str] = Field(None, description="The sys_id of the created/updated variable")
details: Optional[Dict[str, Any]] = Field(None, description="Additional details about the variable")
class ListCatalogItemVariablesParams(BaseModel):
"""Parameters for listing catalog item variables."""
catalog_item_id: str = Field(..., description="The sys_id of the catalog item")
include_details: bool = Field(True, description="Whether to include detailed information about each variable")
limit: Optional[int] = Field(None, description="Maximum number of variables to return")
offset: Optional[int] = Field(None, description="Offset for pagination")
class ListCatalogItemVariablesResponse(BaseModel):
"""Response from listing catalog item variables."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
variables: List[Dict[str, Any]] = Field([], description="List of variables")
count: int = Field(0, description="Total number of variables found")
class UpdateCatalogItemVariableParams(BaseModel):
"""Parameters for updating a catalog item variable."""
variable_id: str = Field(..., description="The sys_id of the variable to update")
label: Optional[str] = Field(None, description="The display label for the variable")
mandatory: Optional[bool] = Field(None, description="Whether the variable is required")
help_text: Optional[str] = Field(None, description="Help text to display with the variable")
default_value: Optional[str] = Field(None, description="Default value for the variable")
description: Optional[str] = Field(None, description="Description of the variable")
order: Optional[int] = Field(None, description="Display order of the variable")
reference_qualifier: Optional[str] = Field(None, description="For reference fields, the query to filter reference options")
max_length: Optional[int] = Field(None, description="Maximum length for string fields")
min: Optional[int] = Field(None, description="Minimum value for numeric fields")
max: Optional[int] = Field(None, description="Maximum value for numeric fields")
def create_catalog_item_variable(
config: ServerConfig,
auth_manager: AuthManager,
params: CreateCatalogItemVariableParams,
) -> CatalogItemVariableResponse:
"""
Create a new variable (form field) for a catalog item.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for creating a catalog item variable.
Returns:
Response with information about the created variable.
"""
api_url = f"{config.instance_url}/api/now/table/item_option_new"
# Build request data
data = {
"cat_item": params.catalog_item_id,
"name": params.name,
"type": params.type,
"question_text": params.label,
"mandatory": str(params.mandatory).lower(), # ServiceNow expects "true"/"false" strings
}
if params.help_text:
data["help_text"] = params.help_text
if params.default_value:
data["default_value"] = params.default_value
if params.description:
data["description"] = params.description
if params.order is not None:
data["order"] = params.order
if params.reference_table:
data["reference"] = params.reference_table
if params.reference_qualifier:
data["reference_qual"] = params.reference_qualifier
if params.max_length:
data["max_length"] = params.max_length
if params.min is not None:
data["min"] = params.min
if params.max is not None:
data["max"] = params.max
# 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 CatalogItemVariableResponse(
success=True,
message="Catalog item variable created successfully",
variable_id=result.get("sys_id"),
details=result,
)
except requests.RequestException as e:
logger.error(f"Failed to create catalog item variable: {e}")
return CatalogItemVariableResponse(
success=False,
message=f"Failed to create catalog item variable: {str(e)}",
)
def list_catalog_item_variables(
config: ServerConfig,
auth_manager: AuthManager,
params: ListCatalogItemVariablesParams,
) -> ListCatalogItemVariablesResponse:
"""
List all variables (form fields) for a catalog item.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for listing catalog item variables.
Returns:
Response with a list of variables for the catalog item.
"""
# Build query parameters
query_params = {
"sysparm_query": f"cat_item={params.catalog_item_id}^ORDERBYorder",
}
if params.limit:
query_params["sysparm_limit"] = params.limit
if params.offset:
query_params["sysparm_offset"] = params.offset
# Include all fields if detailed info is requested
if params.include_details:
query_params["sysparm_display_value"] = "true"
query_params["sysparm_exclude_reference_link"] = "false"
else:
query_params["sysparm_fields"] = "sys_id,name,type,question_text,order,mandatory"
api_url = f"{config.instance_url}/api/now/table/item_option_new"
# Make request
try:
response = requests.get(
api_url,
params=query_params,
headers=auth_manager.get_headers(),
timeout=config.timeout,
)
response.raise_for_status()
result = response.json().get("result", [])
return ListCatalogItemVariablesResponse(
success=True,
message=f"Retrieved {len(result)} variables for catalog item",
variables=result,
count=len(result),
)
except requests.RequestException as e:
logger.error(f"Failed to list catalog item variables: {e}")
return ListCatalogItemVariablesResponse(
success=False,
message=f"Failed to list catalog item variables: {str(e)}",
)
def update_catalog_item_variable(
config: ServerConfig,
auth_manager: AuthManager,
params: UpdateCatalogItemVariableParams,
) -> CatalogItemVariableResponse:
"""
Update an existing variable (form field) for a catalog item.
Args:
config: Server configuration.
auth_manager: Authentication manager.
params: Parameters for updating a catalog item variable.
Returns:
Response with information about the updated variable.
"""
api_url = f"{config.instance_url}/api/now/table/item_option_new/{params.variable_id}"
# Build request data with only parameters that are provided
data = {}
if params.label is not None:
data["question_text"] = params.label
if params.mandatory is not None:
data["mandatory"] = str(params.mandatory).lower() # ServiceNow expects "true"/"false" strings
if params.help_text is not None:
data["help_text"] = params.help_text
if params.default_value is not None:
data["default_value"] = params.default_value
if params.description is not None:
data["description"] = params.description
if params.order is not None:
data["order"] = params.order
if params.reference_qualifier is not None:
data["reference_qual"] = params.reference_qualifier
if params.max_length is not None:
data["max_length"] = params.max_length
if params.min is not None:
data["min"] = params.min
if params.max is not None:
data["max"] = params.max
# If no fields to update, return early
if not data:
return CatalogItemVariableResponse(
success=False,
message="No update parameters provided",
)
# 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 CatalogItemVariableResponse(
success=True,
message="Catalog item variable updated successfully",
variable_id=params.variable_id,
details=result,
)
except requests.RequestException as e:
logger.error(f"Failed to update catalog item variable: {e}")
return CatalogItemVariableResponse(
success=False,
message=f"Failed to update catalog item variable: {str(e)}",
)
```
--------------------------------------------------------------------------------
/tests/test_change_tools.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the change management tools.
"""
import unittest
from unittest.mock import MagicMock, patch
import requests
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.tools.change_tools import (
create_change_request,
list_change_requests,
)
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
class TestChangeTools(unittest.TestCase):
"""Tests for the change management tools."""
def setUp(self):
"""Set up test fixtures."""
self.auth_config = AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(username="test_user", password="test_password"),
)
self.server_config = ServerConfig(
instance_url="https://test.service-now.com",
auth=self.auth_config,
)
self.auth_manager = AuthManager(self.auth_config)
@patch("servicenow_mcp.tools.change_tools.requests.get")
def test_list_change_requests_success(self, mock_get):
"""Test listing change requests successfully."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "change123",
"number": "CHG0010001",
"short_description": "Test Change",
"type": "normal",
"state": "open",
},
{
"sys_id": "change456",
"number": "CHG0010002",
"short_description": "Another Test Change",
"type": "emergency",
"state": "in progress",
},
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = {
"limit": 10,
"timeframe": "upcoming",
}
result = list_change_requests(self.auth_manager, self.server_config, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(len(result["change_requests"]), 2)
self.assertEqual(result["count"], 2)
self.assertEqual(result["total"], 2)
self.assertEqual(result["change_requests"][0]["sys_id"], "change123")
self.assertEqual(result["change_requests"][1]["sys_id"], "change456")
@patch("servicenow_mcp.tools.change_tools.requests.get")
def test_list_change_requests_empty_result(self, mock_get):
"""Test listing change requests with empty result."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {"result": []}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = {
"limit": 10,
"timeframe": "upcoming",
}
result = list_change_requests(self.auth_manager, self.server_config, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(len(result["change_requests"]), 0)
self.assertEqual(result["count"], 0)
self.assertEqual(result["total"], 0)
@patch("servicenow_mcp.tools.change_tools.requests.get")
def test_list_change_requests_missing_result(self, mock_get):
"""Test listing change requests with missing result key."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {} # No "result" key
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = {
"limit": 10,
"timeframe": "upcoming",
}
result = list_change_requests(self.auth_manager, self.server_config, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(len(result["change_requests"]), 0)
self.assertEqual(result["count"], 0)
self.assertEqual(result["total"], 0)
@patch("servicenow_mcp.tools.change_tools.requests.get")
def test_list_change_requests_error(self, mock_get):
"""Test listing change requests with error."""
# Mock the response
mock_get.side_effect = requests.exceptions.RequestException("Test error")
# Call the function
params = {
"limit": 10,
"timeframe": "upcoming",
}
result = list_change_requests(self.auth_manager, self.server_config, params)
# Verify the result
self.assertFalse(result["success"])
self.assertIn("Error listing change requests", result["message"])
@patch("servicenow_mcp.tools.change_tools.requests.get")
def test_list_change_requests_with_filters(self, mock_get):
"""Test listing change requests with filters."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "change123",
"number": "CHG0010001",
"short_description": "Test Change",
"type": "normal",
"state": "open",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function with filters
params = {
"limit": 10,
"state": "open",
"type": "normal",
"category": "Hardware",
"assignment_group": "IT Support",
"timeframe": "upcoming",
"query": "short_description=Test",
}
result = list_change_requests(self.auth_manager, self.server_config, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(len(result["change_requests"]), 1)
# Verify that the correct query parameters were passed to the request
args, kwargs = mock_get.call_args
self.assertIn("params", kwargs)
self.assertIn("sysparm_query", kwargs["params"])
query = kwargs["params"]["sysparm_query"]
# Check that all filters are in the query
self.assertIn("state=open", query)
self.assertIn("type=normal", query)
self.assertIn("category=Hardware", query)
self.assertIn("assignment_group=IT Support", query)
self.assertIn("short_description=Test", query)
# The timeframe filter adds a date comparison, which is harder to test exactly
@patch("servicenow_mcp.tools.change_tools.requests.post")
def test_create_change_request_with_swapped_parameters(self, mock_post):
"""Test creating a change request with swapped parameters (server_config used as auth_manager)."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "change123",
"number": "CHG0010001",
"short_description": "Test Change",
"type": "normal",
}
}
mock_response.raise_for_status = MagicMock()
mock_post.return_value = mock_response
# Create a server_config with a get_headers method to simulate what might happen in Claude Desktop
server_config_with_headers = MagicMock()
server_config_with_headers.instance_url = "https://test.service-now.com"
server_config_with_headers.get_headers.return_value = {"Authorization": "Basic dGVzdF91c2VyOnRlc3RfcGFzc3dvcmQ="}
# Call the function with swapped parameters (server_config as auth_manager)
params = {
"short_description": "Test Change",
"type": "normal",
"risk": "low",
"impact": "medium",
}
result = create_change_request(server_config_with_headers, self.auth_manager, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(result["change_request"]["sys_id"], "change123")
self.assertEqual(result["change_request"]["number"], "CHG0010001")
@patch("servicenow_mcp.tools.change_tools.requests.post")
def test_create_change_request_with_serverconfig_no_get_headers(self, mock_post):
"""Test creating a change request with ServerConfig object that doesn't have get_headers method."""
# This test simulates the exact error we're seeing in Claude Desktop
# Create params for the change request
params = {
"short_description": "Test Change",
"type": "normal",
"risk": "low",
"impact": "medium",
}
# Create a real ServerConfig object (which doesn't have get_headers method)
# and a mock AuthManager object (which doesn't have instance_url)
real_server_config = ServerConfig(
instance_url="https://test.service-now.com",
auth=self.auth_config,
)
mock_auth_manager = MagicMock()
# Explicitly remove get_headers method to simulate the error
if hasattr(mock_auth_manager, 'get_headers'):
delattr(mock_auth_manager, 'get_headers')
# Call the function with parameters that will cause the error
result = create_change_request(real_server_config, mock_auth_manager, params)
# The function should detect the issue and return an error message
self.assertFalse(result["success"])
self.assertIn("Cannot find get_headers method", result["message"])
# Verify that the post method was never called
mock_post.assert_not_called()
@patch("servicenow_mcp.tools.change_tools.requests.post")
def test_create_change_request_with_swapped_parameters_real(self, mock_post):
"""Test creating a change request with swapped parameters (auth_manager and server_config)."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "change123",
"number": "CHG0010001",
"short_description": "Test Change",
"type": "normal",
}
}
mock_response.raise_for_status = MagicMock()
mock_post.return_value = mock_response
# Create params for the change request
params = {
"short_description": "Test Change",
"type": "normal",
"risk": "low",
"impact": "medium",
}
# Call the function with swapped parameters (server_config as first parameter, auth_manager as second)
result = create_change_request(self.server_config, self.auth_manager, params)
# The function should still work correctly
self.assertTrue(result["success"])
self.assertEqual(result["change_request"]["sys_id"], "change123")
self.assertEqual(result["change_request"]["number"], "CHG0010001")
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/tests/test_catalog_resources.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the ServiceNow MCP catalog resources.
"""
import unittest
from unittest.mock import MagicMock, patch
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.resources.catalog import (
CatalogCategoryListParams,
CatalogItemVariableModel,
CatalogListParams,
CatalogResource,
)
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
class TestCatalogResource(unittest.TestCase):
"""Test cases for the catalog resource."""
def setUp(self):
"""Set up test fixtures."""
# Create a mock server config
self.config = ServerConfig(
instance_url="https://example.service-now.com",
auth=AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(username="admin", password="password"),
),
)
# Create a mock auth manager
self.auth_manager = MagicMock(spec=AuthManager)
self.auth_manager.get_headers.return_value = {"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}
# Create the resource
self.resource = CatalogResource(self.config, self.auth_manager)
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_list_catalog_items(self, mock_get):
"""Test listing catalog items."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "item1",
"name": "Laptop",
"short_description": "Request a new laptop",
"category": "Hardware",
"price": "1000",
"picture": "laptop.jpg",
"active": "true",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the method
params = CatalogListParams(
limit=10,
offset=0,
category="Hardware",
query="laptop",
)
result = await self.resource.list_catalog_items(params)
# Check the result
self.assertEqual(len(result), 1)
self.assertEqual(result[0].name, "Laptop")
self.assertEqual(result[0].category, "Hardware")
self.assertTrue(result[0].active)
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item")
self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
self.assertIn("sysparm_query", kwargs["params"])
self.assertIn("active=true", kwargs["params"]["sysparm_query"])
self.assertIn("category=Hardware", kwargs["params"]["sysparm_query"])
self.assertIn("short_descriptionLIKElaptop^ORnameLIKElaptop", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_list_catalog_items_error(self, mock_get):
"""Test listing catalog items with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = Exception("Error")
# Call the method
params = CatalogListParams(
limit=10,
offset=0,
)
result = await self.resource.list_catalog_items(params)
# Check the result
self.assertEqual(len(result), 0)
@patch("servicenow_mcp.resources.catalog.CatalogResource.get_catalog_item_variables")
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_get_catalog_item(self, mock_get, mock_get_variables):
"""Test getting a specific catalog item."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "item1",
"name": "Laptop",
"short_description": "Request a new laptop",
"description": "Request a new laptop for work",
"category": "Hardware",
"price": "1000",
"picture": "laptop.jpg",
"active": "true",
"order": "100",
"delivery_time": "3 days",
"availability": "In Stock",
}
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Mock the variables
mock_get_variables.return_value = [
CatalogItemVariableModel(
sys_id="var1",
name="model",
label="Laptop Model",
type="string",
mandatory=True,
default_value="MacBook Pro",
help_text="Select the laptop model",
order=100,
)
]
# Call the method
result = await self.resource.get_catalog_item("item1")
# Check the result
self.assertEqual(result["sys_id"], "item1")
self.assertEqual(result["name"], "Laptop")
self.assertEqual(result["category"], "Hardware")
self.assertTrue(result["active"])
self.assertEqual(len(result["variables"]), 1)
self.assertEqual(result["variables"][0].name, "model")
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item/item1")
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_get_catalog_item_not_found(self, mock_get):
"""Test getting a catalog item that doesn't exist."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {"result": {}}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the method
result = await self.resource.get_catalog_item("nonexistent")
# Check the result
self.assertIn("error", result)
self.assertIn("not found", result["error"])
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_get_catalog_item_error(self, mock_get):
"""Test getting a catalog item with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = Exception("Error")
# Call the method
result = await self.resource.get_catalog_item("item1")
# Check the result
self.assertIn("error", result)
self.assertIn("Error", result["error"])
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_get_catalog_item_variables(self, mock_get):
"""Test getting variables for a catalog item."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "var1",
"name": "model",
"question_text": "Laptop Model",
"type": "string",
"mandatory": "true",
"default_value": "MacBook Pro",
"help_text": "Select the laptop model",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the method
result = await self.resource.get_catalog_item_variables("item1")
# Check the result
self.assertEqual(len(result), 1)
self.assertEqual(result[0].name, "model")
self.assertEqual(result[0].label, "Laptop Model")
self.assertEqual(result[0].type, "string")
self.assertTrue(result[0].mandatory)
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/item_option_new")
self.assertEqual(kwargs["params"]["sysparm_query"], "cat_item=item1^ORDERBYorder")
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_get_catalog_item_variables_error(self, mock_get):
"""Test getting variables for a catalog item with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = Exception("Error")
# Call the method
result = await self.resource.get_catalog_item_variables("item1")
# Check the result
self.assertEqual(len(result), 0)
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_list_catalog_categories(self, mock_get):
"""Test listing catalog categories."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "cat1",
"title": "Hardware",
"description": "Hardware requests",
"parent": "",
"icon": "hardware.png",
"active": "true",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the method
params = CatalogCategoryListParams(
limit=10,
offset=0,
query="hardware",
)
result = await self.resource.list_catalog_categories(params)
# Check the result
self.assertEqual(len(result), 1)
self.assertEqual(result[0].title, "Hardware")
self.assertEqual(result[0].description, "Hardware requests")
self.assertTrue(result[0].active)
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_category")
self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
self.assertIn("sysparm_query", kwargs["params"])
self.assertIn("active=true", kwargs["params"]["sysparm_query"])
self.assertIn("titleLIKEhardware^ORdescriptionLIKEhardware", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.resources.catalog.requests.get")
async def test_list_catalog_categories_error(self, mock_get):
"""Test listing catalog categories with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = Exception("Error")
# Call the method
params = CatalogCategoryListParams(
limit=10,
offset=0,
)
result = await self.resource.list_catalog_categories(params)
# Check the result
self.assertEqual(len(result), 0)
@patch("servicenow_mcp.resources.catalog.CatalogResource.get_catalog_item")
async def test_read(self, mock_get_catalog_item):
"""Test reading a catalog item."""
# Mock the get_catalog_item method
mock_get_catalog_item.return_value = {
"sys_id": "item1",
"name": "Laptop",
}
# Call the method
result = await self.resource.read({"item_id": "item1"})
# Check the result
self.assertEqual(result["sys_id"], "item1")
self.assertEqual(result["name"], "Laptop")
# Check that the correct method was called
mock_get_catalog_item.assert_called_once_with("item1")
async def test_read_missing_param(self):
"""Test reading a catalog item with missing parameter."""
# Call the method
result = await self.resource.read({})
# Check the result
self.assertIn("error", result)
self.assertIn("Missing item_id parameter", result["error"])
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/tests/test_catalog_variables.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the catalog item variables tools.
"""
import unittest
from unittest.mock import MagicMock, patch
import requests
from servicenow_mcp.tools.catalog_variables import (
CreateCatalogItemVariableParams,
ListCatalogItemVariablesParams,
UpdateCatalogItemVariableParams,
create_catalog_item_variable,
list_catalog_item_variables,
update_catalog_item_variable,
)
from servicenow_mcp.utils.config import ServerConfig, AuthConfig, AuthType, BasicAuthConfig
class TestCatalogVariablesTools(unittest.TestCase):
"""
Test the catalog item variables tools.
"""
def setUp(self):
"""Set up the test environment."""
self.config = ServerConfig(
instance_url="https://test.service-now.com",
timeout=10,
auth=AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(
username="test_user",
password="test_password"
)
),
)
self.auth_manager = MagicMock()
self.auth_manager.get_headers.return_value = {"Content-Type": "application/json"}
@patch("requests.post")
def test_create_catalog_item_variable(self, mock_post):
"""Test create_catalog_item_variable function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "abc123",
"name": "test_variable",
"type": "string",
"question_text": "Test Variable",
"mandatory": "false",
}
}
mock_post.return_value = mock_response
# Create test params
params = CreateCatalogItemVariableParams(
catalog_item_id="item123",
name="test_variable",
type="string",
label="Test Variable",
mandatory=False,
)
# Call function
result = create_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.variable_id, "abc123")
self.assertIsNotNone(result.details)
# Verify mock was called correctly
mock_post.assert_called_once()
call_args = mock_post.call_args
self.assertEqual(
call_args[0][0], f"{self.config.instance_url}/api/now/table/item_option_new"
)
self.assertEqual(call_args[1]["json"]["cat_item"], "item123")
self.assertEqual(call_args[1]["json"]["name"], "test_variable")
self.assertEqual(call_args[1]["json"]["type"], "string")
self.assertEqual(call_args[1]["json"]["question_text"], "Test Variable")
self.assertEqual(call_args[1]["json"]["mandatory"], "false")
@patch("requests.post")
def test_create_catalog_item_variable_with_optional_params(self, mock_post):
"""Test create_catalog_item_variable function with optional parameters."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "abc123",
"name": "test_variable",
"type": "reference",
"question_text": "Test Reference",
"mandatory": "true",
"reference": "sys_user",
"reference_qual": "active=true",
"help_text": "Select a user",
"default_value": "admin",
"description": "Reference to a user",
"order": 100,
}
}
mock_post.return_value = mock_response
# Create test params with optional fields
params = CreateCatalogItemVariableParams(
catalog_item_id="item123",
name="test_variable",
type="reference",
label="Test Reference",
mandatory=True,
help_text="Select a user",
default_value="admin",
description="Reference to a user",
order=100,
reference_table="sys_user",
reference_qualifier="active=true",
)
# Call function
result = create_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.variable_id, "abc123")
# Verify mock was called correctly
mock_post.assert_called_once()
call_args = mock_post.call_args
self.assertEqual(call_args[1]["json"]["reference"], "sys_user")
self.assertEqual(call_args[1]["json"]["reference_qual"], "active=true")
self.assertEqual(call_args[1]["json"]["help_text"], "Select a user")
self.assertEqual(call_args[1]["json"]["default_value"], "admin")
self.assertEqual(call_args[1]["json"]["description"], "Reference to a user")
self.assertEqual(call_args[1]["json"]["order"], 100)
@patch("requests.post")
def test_create_catalog_item_variable_error(self, mock_post):
"""Test create_catalog_item_variable function with error."""
# Configure mock to raise exception
mock_post.side_effect = requests.RequestException("Test error")
# Create test params
params = CreateCatalogItemVariableParams(
catalog_item_id="item123",
name="test_variable",
type="string",
label="Test Variable",
)
# Call function
result = create_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result
self.assertFalse(result.success)
self.assertTrue("failed" in result.message.lower())
@patch("requests.get")
def test_list_catalog_item_variables(self, mock_get):
"""Test list_catalog_item_variables function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "var1",
"name": "variable1",
"type": "string",
"question_text": "Variable 1",
"order": 100,
"mandatory": "true",
},
{
"sys_id": "var2",
"name": "variable2",
"type": "integer",
"question_text": "Variable 2",
"order": 200,
"mandatory": "false",
},
]
}
mock_get.return_value = mock_response
# Create test params
params = ListCatalogItemVariablesParams(
catalog_item_id="item123",
include_details=True,
)
# Call function
result = list_catalog_item_variables(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.count, 2)
self.assertEqual(len(result.variables), 2)
self.assertEqual(result.variables[0]["sys_id"], "var1")
self.assertEqual(result.variables[1]["sys_id"], "var2")
# Verify mock was called correctly
mock_get.assert_called_once()
call_args = mock_get.call_args
self.assertEqual(
call_args[0][0], f"{self.config.instance_url}/api/now/table/item_option_new"
)
self.assertEqual(
call_args[1]["params"]["sysparm_query"], "cat_item=item123^ORDERBYorder"
)
self.assertEqual(call_args[1]["params"]["sysparm_display_value"], "true")
self.assertEqual(call_args[1]["params"]["sysparm_exclude_reference_link"], "false")
@patch("requests.get")
def test_list_catalog_item_variables_with_pagination(self, mock_get):
"""Test list_catalog_item_variables function with pagination parameters."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {"result": [{"sys_id": "var1"}]}
mock_get.return_value = mock_response
# Create test params with pagination
params = ListCatalogItemVariablesParams(
catalog_item_id="item123",
include_details=False,
limit=10,
offset=20,
)
# Call function
result = list_catalog_item_variables(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
# Verify mock was called correctly with pagination
mock_get.assert_called_once()
call_args = mock_get.call_args
self.assertEqual(call_args[1]["params"]["sysparm_limit"], 10)
self.assertEqual(call_args[1]["params"]["sysparm_offset"], 20)
self.assertEqual(
call_args[1]["params"]["sysparm_fields"],
"sys_id,name,type,question_text,order,mandatory",
)
@patch("requests.get")
def test_list_catalog_item_variables_error(self, mock_get):
"""Test list_catalog_item_variables function with error."""
# Configure mock to raise exception
mock_get.side_effect = requests.RequestException("Test error")
# Create test params
params = ListCatalogItemVariablesParams(
catalog_item_id="item123",
)
# Call function
result = list_catalog_item_variables(self.config, self.auth_manager, params)
# Verify result
self.assertFalse(result.success)
self.assertTrue("failed" in result.message.lower())
@patch("requests.patch")
def test_update_catalog_item_variable(self, mock_patch):
"""Test update_catalog_item_variable function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "var1",
"question_text": "Updated Variable",
"mandatory": "true",
"help_text": "This is help text",
}
}
mock_patch.return_value = mock_response
# Create test params
params = UpdateCatalogItemVariableParams(
variable_id="var1",
label="Updated Variable",
mandatory=True,
help_text="This is help text",
)
# Call function
result = update_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.variable_id, "var1")
self.assertIsNotNone(result.details)
# Verify mock was called correctly
mock_patch.assert_called_once()
call_args = mock_patch.call_args
self.assertEqual(
call_args[0][0],
f"{self.config.instance_url}/api/now/table/item_option_new/var1",
)
self.assertEqual(call_args[1]["json"]["question_text"], "Updated Variable")
self.assertEqual(call_args[1]["json"]["mandatory"], "true")
self.assertEqual(call_args[1]["json"]["help_text"], "This is help text")
@patch("requests.patch")
def test_update_catalog_item_variable_no_params(self, mock_patch):
"""Test update_catalog_item_variable function with no update parameters."""
# Create test params with no updates (only ID)
params = UpdateCatalogItemVariableParams(
variable_id="var1",
)
# Call function
result = update_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result - should fail since no update parameters provided
self.assertFalse(result.success)
self.assertEqual(result.message, "No update parameters provided")
# Verify mock was not called
mock_patch.assert_not_called()
@patch("requests.patch")
def test_update_catalog_item_variable_error(self, mock_patch):
"""Test update_catalog_item_variable function with error."""
# Configure mock to raise exception
mock_patch.side_effect = requests.RequestException("Test error")
# Create test params
params = UpdateCatalogItemVariableParams(
variable_id="var1",
label="Updated Variable",
)
# Call function
result = update_catalog_item_variable(self.config, self.auth_manager, params)
# Verify result
self.assertFalse(result.success)
self.assertTrue("failed" in result.message.lower())
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/server.py:
--------------------------------------------------------------------------------
```python
"""
ServiceNow MCP Server
This module provides the main implementation of the ServiceNow MCP server.
"""
import json
import logging
import os
from typing import Any, Dict, List, Union
import mcp.types as types
import yaml
from mcp.server.lowlevel import Server
from pydantic import ValidationError
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.tools.knowledge_base import (
create_category as create_kb_category_tool,
)
from servicenow_mcp.tools.knowledge_base import (
list_categories as list_kb_categories_tool,
)
from servicenow_mcp.utils.config import ServerConfig
from servicenow_mcp.utils.tool_utils import get_tool_definitions
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Define path for the configuration file
TOOL_PACKAGE_CONFIG_PATH = os.getenv("TOOL_PACKAGE_CONFIG_PATH", "config/tool_packages.yaml")
def serialize_tool_output(result: Any, tool_name: str) -> str:
"""Serializes tool output to a string, preferably JSON indented."""
try:
if isinstance(result, str):
# If it's already a string, assume it's intended as such
# Try to parse/re-dump JSON for consistent formatting if it looks like JSON
try:
parsed = json.loads(result)
return json.dumps(parsed, indent=2)
except json.JSONDecodeError:
return result # Return as is if not valid JSON
elif isinstance(result, dict):
# Dump dicts to JSON
return json.dumps(result, indent=2)
elif hasattr(result, "model_dump_json"): # Pydantic v2
# Prefer Pydantic v2 model_dump_json
# The indent argument might not be supported by all versions/models,
# so we might need to dump to dict first if indent is crucial.
try:
return result.model_dump_json(indent=2)
except TypeError: # Handle case where indent is not supported
return json.dumps(result.model_dump(), indent=2)
elif hasattr(result, "model_dump"): # Pydantic v2 fallback
# Fallback to Pydantic v2 model_dump -> dict -> json
return json.dumps(result.model_dump(), indent=2)
elif hasattr(result, "dict"): # Pydantic v1
# Fallback to Pydantic v1 dict -> json
return json.dumps(result.dict(), indent=2)
else:
# Absolute fallback: convert to string
logger.warning(
f"Could not serialize result for tool '{tool_name}' to JSON, falling back to str(). Type: {type(result)}"
)
return str(result)
except Exception as e:
logger.error(f"Error during serialization for tool '{tool_name}': {e}", exc_info=True)
# Return an error message string formatted as JSON
return json.dumps(
{"error": f"Serialization failed for tool {tool_name}", "details": str(e)}, indent=2
)
class ServiceNowMCP:
"""
ServiceNow MCP Server implementation.
This class provides a Model Context Protocol (MCP) server for ServiceNow,
allowing LLMs to interact with ServiceNow data and functionality.
It supports loading specific tool packages via the MCP_TOOL_PACKAGE env var.
"""
def __init__(self, config: Union[Dict, ServerConfig]):
"""
Initialize the ServiceNow MCP server.
Args:
config: Server configuration, either as a dictionary or ServerConfig object.
"""
if isinstance(config, dict):
self.config = ServerConfig(**config)
else:
self.config = config
self.auth_manager = AuthManager(self.config.auth, self.config.instance_url)
self.mcp_server = Server("ServiceNow") # Use low-level Server
self.name = "ServiceNow"
self.package_definitions: Dict[str, List[str]] = {}
self.enabled_tool_names: List[str] = []
self.current_package_name: str = "none"
self._load_package_config()
self._determine_enabled_tools()
# Get tool definitions, passing the aliased KB tool functions if needed
self.tool_definitions = get_tool_definitions(
create_kb_category_tool, list_kb_categories_tool
)
self._register_handlers()
def _register_handlers(self):
"""Register the list_tools and call_tool handlers."""
self.mcp_server.list_tools()(self._list_tools_impl)
self.mcp_server.call_tool()(self._call_tool_impl)
logger.info("Registered list_tools and call_tool handlers.")
def _load_package_config(self):
"""Load tool package definitions from the YAML configuration file."""
config_path = TOOL_PACKAGE_CONFIG_PATH
if not os.path.isabs(config_path):
config_path = os.path.join(os.path.dirname(__file__), "..", "..", config_path)
config_path = os.path.abspath(config_path)
try:
with open(config_path, "r") as f:
loaded_config = yaml.safe_load(f)
if isinstance(loaded_config, dict):
self.package_definitions = loaded_config
logger.info(f"Successfully loaded tool package config from {config_path}")
else:
logger.error(
f"Invalid format in {config_path}: Expected a dictionary, got {type(loaded_config)}. No packages loaded."
)
self.package_definitions = {}
except FileNotFoundError:
logger.error(
f"Tool package config file not found at {config_path}. No packages loaded."
)
self.package_definitions = {}
except yaml.YAMLError as e:
logger.error(
f"Error parsing tool package config file {config_path}: {e}. No packages loaded."
)
self.package_definitions = {}
except Exception as e:
logger.error(f"Unexpected error loading tool package config {config_path}: {e}")
self.package_definitions = {}
def _determine_enabled_tools(self):
"""Determine which tool package and tools to enable based on environment variable."""
requested_package = os.getenv("MCP_TOOL_PACKAGE", "full").strip()
if not requested_package:
self.current_package_name = "full"
logger.info("MCP_TOOL_PACKAGE is empty, defaulting to 'full' package.")
elif requested_package in self.package_definitions:
self.current_package_name = requested_package
logger.info(f"MCP_TOOL_PACKAGE set to '{self.current_package_name}'.")
else:
self.current_package_name = "none"
logger.warning(
f"MCP_TOOL_PACKAGE '{requested_package}' is not a valid package name. "
f"Valid packages: {list(self.package_definitions.keys())}. Loading 'none' package."
)
if self.package_definitions:
self.enabled_tool_names = self.package_definitions.get(self.current_package_name, [])
else:
self.enabled_tool_names = []
logger.info(
f"Loading package '{self.current_package_name}' with {len(self.enabled_tool_names)} tools."
)
async def _list_tools_impl(self) -> List[types.Tool]:
"""Implementation for the list_tools MCP endpoint."""
tool_list: List[types.Tool] = []
# Add the introspection tool if not 'none' package
if self.current_package_name != "none":
tool_list.append(
types.Tool(
name="list_tool_packages",
description="Lists available tool packages and the currently loaded one.",
inputSchema={
"type": "object",
"properties": {
"random_string": {
"type": "string",
"description": "Dummy parameter for no-parameter tools",
}
},
"required": ["random_string"],
},
)
)
# Iterate through defined tools and add enabled ones
for tool_name, definition in self.tool_definitions.items():
if tool_name in self.enabled_tool_names:
_impl_func, params_model, _return_annotation, description, _serialization = (
definition
)
try:
schema = params_model.model_json_schema()
tool_list.append(
types.Tool(name=tool_name, description=description, inputSchema=schema)
)
except Exception as e:
logger.error(
f"Failed to generate schema for tool '{tool_name}': {e}", exc_info=True
)
logger.debug(f"Listing {len(tool_list)} tools for package '{self.current_package_name}'.")
return tool_list
async def _call_tool_impl(self, name: str, arguments: dict) -> list[types.TextContent]:
"""
Implementation for the call_tool MCP endpoint.
Handles argument parsing, tool execution, result serialization (to string),
and returning a list containing a single TextContent object.
Args:
name: The name of the tool to call.
arguments: The arguments for the tool as a dictionary.
Returns:
A list containing a single TextContent object with the tool output.
Raises:
ValueError: If the tool is unknown, disabled, or if arguments are invalid.
RuntimeError: If tool execution or serialization fails.
"""
logger.info(f"Received call_tool request for tool '{name}'")
# Handle the introspection tool separately
if name == "list_tool_packages":
if self.current_package_name == "none":
raise ValueError(
"Tool 'list_tool_packages' is not available in the 'none' package."
)
result_dict = self._list_tool_packages_impl()
serialized_string = json.dumps(result_dict, indent=2)
# Return a list with a TextContent object
return [types.TextContent(type="text", text=serialized_string)]
# Check if the tool exists and is enabled
if name not in self.tool_definitions:
raise ValueError(f"Unknown tool: {name}")
if name not in self.enabled_tool_names:
raise ValueError(
f"Tool '{name}' is not enabled in the current package '{self.current_package_name}'."
)
# Get tool definition (we don't need the serialization hint anymore)
definition = self.tool_definitions[name]
impl_func, params_model, _return_annotation, _description, _serialization = definition
# Validate and parse arguments using the Pydantic model
try:
params = params_model(**arguments)
logger.debug(f"Parsed arguments for tool '{name}': {params}")
except ValidationError as e:
logger.error(f"Invalid arguments for tool '{name}': {e}", exc_info=True)
raise ValueError(f"Invalid arguments for tool '{name}': {e}") from e
except Exception as e:
logger.error(
f"Unexpected error parsing arguments for tool '{name}': {e}", exc_info=True
)
raise ValueError(f"Failed to parse arguments for tool '{name}': {e}")
# Execute the tool implementation function
try:
result = impl_func(self.config, self.auth_manager, params)
logger.debug(f"Raw result type from tool '{name}': {type(result)}")
except Exception as e:
logger.error(f"Error executing tool '{name}': {e}", exc_info=True)
raise RuntimeError(f"Error during execution of tool '{name}': {e}") from e
# Serialize the result to a string (preferably JSON) using the helper
serialized_string = serialize_tool_output(result, name)
logger.debug(f"Serialized value for tool '{name}': {serialized_string[:500]}...")
# Return a list with a TextContent object
return [types.TextContent(type="text", text=serialized_string)]
def _list_tool_packages_impl(self) -> Dict[str, Any]:
"""Implementation logic for the list_tool_packages tool."""
available_packages = list(self.package_definitions.keys())
return {
"current_package": self.current_package_name,
"available_packages": available_packages,
"message": (
f"Currently loaded package: '{self.current_package_name}'. "
f"Set MCP_TOOL_PACKAGE env var to one of {available_packages} to switch."
),
}
def start(self) -> Server:
"""
Prepares and returns the configured low-level MCP Server instance.
The caller (e.g., cli.py) is responsible for obtaining the server
instance from this method and running it within an appropriate
async transport context (e.g., mcp.server.stdio.stdio_server).
Returns:
The configured mcp.server.lowlevel.Server instance.
"""
logger.info(
"ServiceNowMCP instance configured. Returning low-level server instance for external execution."
)
# The actual running of the server (server.run(...)) must happen
# within an async context managed by the caller (e.g., using anyio
# and a specific transport like stdio_server or SseServerTransport).
return self.mcp_server
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/epic_tools.py:
--------------------------------------------------------------------------------
```python
"""
Epic management tools for the ServiceNow MCP server.
This module provides tools for managing epics 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 CreateEpicParams(BaseModel):
"""Parameters for creating an epic."""
short_description: str = Field(..., description="Short description of the epic")
description: Optional[str] = Field(None, description="Detailed description of the epic")
priority: Optional[str] = Field(None, description="Priority of epic (1 is Critical, 2 is High, 3 is Moderate, 4 is Low, 5 is Planning)")
state: Optional[str] = Field(None, description="State of story (-6 is Draft,1 is Ready,2 is Work in progress, 3 is Complete, 4 is Cancelled)")
assignment_group: Optional[str] = Field(None, description="Group assigned to the epic")
assigned_to: Optional[str] = Field(None, description="User assigned to the epic")
work_notes: Optional[str] = Field(None, description="Work notes to add to the epic. Used for adding notes and comments to an epic")
class UpdateEpicParams(BaseModel):
"""Parameters for updating an epic."""
epic_id: str = Field(..., description="Epic ID or sys_id")
short_description: Optional[str] = Field(None, description="Short description of the epic")
description: Optional[str] = Field(None, description="Detailed description of the epic")
priority: Optional[str] = Field(None, description="Priority of epic (1 is Critical, 2 is High, 3 is Moderate, 4 is Low, 5 is Planning)")
state: Optional[str] = Field(None, description="State of story (-6 is Draft,1 is Ready,2 is Work in progress, 3 is Complete, 4 is Cancelled)")
assignment_group: Optional[str] = Field(None, description="Group assigned to the epic")
assigned_to: Optional[str] = Field(None, description="User assigned to the epic")
work_notes: Optional[str] = Field(None, description="Work notes to add to the epic. Used for adding notes and comments to an epic")
class ListEpicsParams(BaseModel):
"""Parameters for listing epics."""
limit: Optional[int] = Field(10, description="Maximum number of records to return")
offset: Optional[int] = Field(0, description="Offset to start from")
priority: Optional[str] = Field(None, description="Filter by priority")
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")
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_epic(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Create a new epic in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for creating the epic.
Returns:
The created epic.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
CreateEpicParams,
required_fields=["short_description"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {
"short_description": validated_params.short_description,
}
# Add optional fields if provided
if validated_params.description:
data["description"] = validated_params.description
if validated_params.priority:
data["priority"] = validated_params.priority
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.assigned_to:
data["assigned_to"] = validated_params.assigned_to
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/rm_epic"
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Epic created successfully",
"epic": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error creating epic: {e}")
return {
"success": False,
"message": f"Error creating epic: {str(e)}",
}
def update_epic(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Update an existing epic in ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for updating the epic.
Returns:
The updated epic.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
UpdateEpicParams,
required_fields=["epic_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {}
# Add optional 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.priority:
data["priority"] = validated_params.priority
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.assigned_to:
data["assigned_to"] = validated_params.assigned_to
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/rm_epic/{validated_params.epic_id}"
try:
response = requests.put(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Epic updated successfully",
"epic": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error updating epic: {e}")
return {
"success": False,
"message": f"Error updating epic: {str(e)}",
}
def list_epics(
auth_manager: AuthManager,
server_config: ServerConfig,
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
List epics from ServiceNow.
Args:
auth_manager: The authentication manager.
server_config: The server configuration.
params: The parameters for listing epics.
Returns:
A list of epics.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
ListEpicsParams
)
if not result["success"]:
return result
validated_params = result["params"]
# Build the query
query_parts = []
if validated_params.priority:
query_parts.append(f"priority={validated_params.priority}")
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/rm_epic"
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
epics = result.get("result", [])
count = len(epics)
return {
"success": True,
"epics": epics,
"count": count,
"total": count, # Use count as total if total is not provided
}
except requests.exceptions.RequestException as e:
logger.error(f"Error listing epics: {e}")
return {
"success": False,
"message": f"Error listing epics: {str(e)}",
}
```
--------------------------------------------------------------------------------
/tests/test_user_tools.py:
--------------------------------------------------------------------------------
```python
"""
Tests for user management tools.
"""
import unittest
from unittest.mock import MagicMock, patch
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.tools.user_tools import (
AddGroupMembersParams,
CreateGroupParams,
CreateUserParams,
GetUserParams,
ListUsersParams,
ListGroupsParams,
RemoveGroupMembersParams,
UpdateGroupParams,
UpdateUserParams,
add_group_members,
create_group,
create_user,
get_user,
list_users,
list_groups,
remove_group_members,
update_group,
update_user,
)
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
class TestUserTools(unittest.TestCase):
"""Tests for user management tools."""
def setUp(self):
"""Set up test environment."""
# Create config and auth manager
self.config = ServerConfig(
instance_url="https://example.service-now.com",
auth=AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(username="admin", password="password"),
),
)
self.auth_manager = AuthManager(self.config.auth)
# Mock auth_manager.get_headers() method
self.auth_manager.get_headers = MagicMock(return_value={"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="})
@patch("requests.post")
def test_create_user(self, mock_post):
"""Test create_user function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "user123",
"user_name": "alice.radiology",
}
}
mock_post.return_value = mock_response
# Create test params
params = CreateUserParams(
user_name="alice.radiology",
first_name="Alice",
last_name="Radiology",
email="[email protected]",
department="Radiology",
title="Doctor",
)
# Call function
result = create_user(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.user_id, "user123")
self.assertEqual(result.user_name, "alice.radiology")
# Verify mock was called correctly
mock_post.assert_called_once()
call_args = mock_post.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user")
self.assertEqual(call_args[1]["json"]["user_name"], "alice.radiology")
self.assertEqual(call_args[1]["json"]["first_name"], "Alice")
self.assertEqual(call_args[1]["json"]["last_name"], "Radiology")
self.assertEqual(call_args[1]["json"]["email"], "[email protected]")
self.assertEqual(call_args[1]["json"]["department"], "Radiology")
self.assertEqual(call_args[1]["json"]["title"], "Doctor")
@patch("requests.patch")
def test_update_user(self, mock_patch):
"""Test update_user function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "user123",
"user_name": "alice.radiology",
}
}
mock_patch.return_value = mock_response
# Create test params
params = UpdateUserParams(
user_id="user123",
manager="user456",
title="Senior Doctor",
)
# Call function
result = update_user(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.user_id, "user123")
self.assertEqual(result.user_name, "alice.radiology")
# Verify mock was called correctly
mock_patch.assert_called_once()
call_args = mock_patch.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user/user123")
self.assertEqual(call_args[1]["json"]["manager"], "user456")
self.assertEqual(call_args[1]["json"]["title"], "Senior Doctor")
@patch("requests.get")
def test_get_user(self, mock_get):
"""Test get_user function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "user123",
"user_name": "alice.radiology",
"first_name": "Alice",
"last_name": "Radiology",
"email": "[email protected]",
}
]
}
mock_get.return_value = mock_response
# Create test params
params = GetUserParams(
user_name="alice.radiology",
)
# Call function
result = get_user(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result["success"])
self.assertEqual(result["user"]["sys_id"], "user123")
self.assertEqual(result["user"]["user_name"], "alice.radiology")
# Verify mock was called correctly
mock_get.assert_called_once()
call_args = mock_get.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user")
self.assertEqual(call_args[1]["params"]["sysparm_query"], "user_name=alice.radiology")
@patch("requests.get")
def test_list_users(self, mock_get):
"""Test list_users function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "user123",
"user_name": "alice.radiology",
},
{
"sys_id": "user456",
"user_name": "bob.chiefradiology",
}
]
}
mock_get.return_value = mock_response
# Create test params
params = ListUsersParams(
department="Radiology",
limit=10,
)
# Call function
result = list_users(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result["success"])
self.assertEqual(len(result["users"]), 2)
self.assertEqual(result["users"][0]["sys_id"], "user123")
self.assertEqual(result["users"][1]["sys_id"], "user456")
# Verify mock was called correctly
mock_get.assert_called_once()
call_args = mock_get.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user")
self.assertEqual(call_args[1]["params"]["sysparm_limit"], "10")
self.assertIn("department=Radiology", call_args[1]["params"]["sysparm_query"])
@patch("requests.get")
def test_list_groups(self, mock_get):
"""Test list_groups function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "group123",
"name": "IT Support",
"description": "IT support team",
"active": "true",
"type": "it"
},
{
"sys_id": "group456",
"name": "HR Team",
"description": "Human Resources team",
"active": "true",
"type": "administrative"
}
]
}
mock_get.return_value = mock_response
# Create test params
params = ListGroupsParams(
active=True,
type="it",
query="support",
limit=10,
)
# Call function
result = list_groups(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result["success"])
self.assertEqual(len(result["groups"]), 2)
self.assertEqual(result["groups"][0]["sys_id"], "group123")
self.assertEqual(result["groups"][1]["sys_id"], "group456")
self.assertEqual(result["count"], 2)
# Verify mock was called correctly
mock_get.assert_called_once()
call_args = mock_get.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user_group")
self.assertEqual(call_args[1]["params"]["sysparm_limit"], "10")
self.assertEqual(call_args[1]["params"]["sysparm_offset"], "0")
self.assertEqual(call_args[1]["params"]["sysparm_display_value"], "true")
self.assertIn("active=true", call_args[1]["params"]["sysparm_query"])
self.assertIn("type=it", call_args[1]["params"]["sysparm_query"])
self.assertIn("nameLIKE", call_args[1]["params"]["sysparm_query"])
self.assertIn("descriptionLIKE", call_args[1]["params"]["sysparm_query"])
@patch("requests.post")
def test_create_group(self, mock_post):
"""Test create_group function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "group123",
"name": "Biomedical Engineering",
}
}
mock_post.return_value = mock_response
# Create test params
params = CreateGroupParams(
name="Biomedical Engineering",
description="Group for biomedical engineering staff",
manager="user456",
)
# Call function
result = create_group(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.group_id, "group123")
self.assertEqual(result.group_name, "Biomedical Engineering")
# Verify mock was called correctly
mock_post.assert_called_once()
call_args = mock_post.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user_group")
self.assertEqual(call_args[1]["json"]["name"], "Biomedical Engineering")
self.assertEqual(call_args[1]["json"]["description"], "Group for biomedical engineering staff")
self.assertEqual(call_args[1]["json"]["manager"], "user456")
@patch("requests.patch")
def test_update_group(self, mock_patch):
"""Test update_group function."""
# Configure mock
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "group123",
"name": "Biomedical Engineering",
}
}
mock_patch.return_value = mock_response
# Create test params
params = UpdateGroupParams(
group_id="group123",
description="Updated description for biomedical engineering group",
manager="user789",
)
# Call function
result = update_group(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.group_id, "group123")
self.assertEqual(result.group_name, "Biomedical Engineering")
# Verify mock was called correctly
mock_patch.assert_called_once()
call_args = mock_patch.call_args
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user_group/group123")
self.assertEqual(call_args[1]["json"]["description"], "Updated description for biomedical engineering group")
self.assertEqual(call_args[1]["json"]["manager"], "user789")
@patch("servicenow_mcp.tools.user_tools.get_user")
@patch("requests.post")
def test_add_group_members(self, mock_post, mock_get_user):
"""Test add_group_members function."""
# Configure mocks
mock_post_response = MagicMock()
mock_post_response.raise_for_status = MagicMock()
mock_post.return_value = mock_post_response
mock_get_user.return_value = {
"success": True,
"message": "User found",
"user": {
"sys_id": "user123",
"user_name": "alice.radiology",
}
}
# Create test params
params = AddGroupMembersParams(
group_id="group123",
members=["alice.radiology", "admin"],
)
# Call function
result = add_group_members(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.group_id, "group123")
# Verify mock was called correctly
self.assertEqual(mock_post.call_count, 2) # Once for each member
call_args = mock_post.call_args_list[0]
self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/sys_user_grmember")
self.assertEqual(call_args[1]["json"]["group"], "group123")
self.assertEqual(call_args[1]["json"]["user"], "user123")
@patch("servicenow_mcp.tools.user_tools.get_user")
@patch("requests.get")
@patch("requests.delete")
def test_remove_group_members(self, mock_delete, mock_get, mock_get_user):
"""Test remove_group_members function."""
# Configure mocks
mock_delete_response = MagicMock()
mock_delete_response.raise_for_status = MagicMock()
mock_delete.return_value = mock_delete_response
mock_get_response = MagicMock()
mock_get_response.raise_for_status = MagicMock()
mock_get_response.json.return_value = {
"result": [
{
"sys_id": "member123",
"user": {
"value": "user123",
"display_value": "Alice Radiology",
},
"group": {
"value": "group123",
"display_value": "Biomedical Engineering",
},
}
]
}
mock_get.return_value = mock_get_response
mock_get_user.return_value = {
"success": True,
"message": "User found",
"user": {
"sys_id": "user123",
"user_name": "alice.radiology",
}
}
# Create test params
params = RemoveGroupMembersParams(
group_id="group123",
members=["alice.radiology"],
)
# Call function
result = remove_group_members(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.group_id, "group123")
# Verify mock was called correctly
mock_get.assert_called_once()
get_call_args = mock_get.call_args
self.assertEqual(get_call_args[0][0], f"{self.config.api_url}/table/sys_user_grmember")
self.assertEqual(get_call_args[1]["params"]["sysparm_query"], "group=group123^user=user123")
mock_delete.assert_called_once()
delete_call_args = mock_delete.call_args
self.assertEqual(delete_call_args[0][0], f"{self.config.api_url}/table/sys_user_grmember/member123")
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/tests/test_catalog_tools.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the ServiceNow MCP catalog tools.
"""
import unittest
from unittest.mock import MagicMock, patch
import requests
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.tools.catalog_tools import (
GetCatalogItemParams,
ListCatalogCategoriesParams,
ListCatalogItemsParams,
CreateCatalogCategoryParams,
UpdateCatalogCategoryParams,
MoveCatalogItemsParams,
get_catalog_item,
get_catalog_item_variables,
list_catalog_categories,
list_catalog_items,
create_catalog_category,
update_catalog_category,
move_catalog_items,
)
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
class TestCatalogTools(unittest.TestCase):
"""Test cases for the catalog tools."""
def setUp(self):
"""Set up test fixtures."""
# Create a mock server config
self.config = ServerConfig(
instance_url="https://example.service-now.com",
auth=AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(username="admin", password="password"),
),
)
# Create a mock auth manager
self.auth_manager = MagicMock(spec=AuthManager)
self.auth_manager.get_headers.return_value = {"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_list_catalog_items(self, mock_get):
"""Test listing catalog items."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "item1",
"name": "Laptop",
"short_description": "Request a new laptop",
"category": "Hardware",
"price": "1000",
"picture": "laptop.jpg",
"active": "true",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = ListCatalogItemsParams(
limit=10,
offset=0,
category="Hardware",
query="laptop",
active=True,
)
result = list_catalog_items(self.config, self.auth_manager, params)
# Check the result
self.assertTrue(result["success"])
self.assertEqual(len(result["items"]), 1)
self.assertEqual(result["items"][0]["name"], "Laptop")
self.assertEqual(result["items"][0]["category"], "Hardware")
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item")
self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
self.assertIn("sysparm_query", kwargs["params"])
self.assertIn("active=true", kwargs["params"]["sysparm_query"])
self.assertIn("category=Hardware", kwargs["params"]["sysparm_query"])
self.assertIn("short_descriptionLIKElaptop^ORnameLIKElaptop", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_list_catalog_items_error(self, mock_get):
"""Test listing catalog items with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = requests.exceptions.RequestException("Error")
# Call the function
params = ListCatalogItemsParams(
limit=10,
offset=0,
)
result = list_catalog_items(self.config, self.auth_manager, params)
# Check the result
self.assertFalse(result["success"])
self.assertEqual(len(result["items"]), 0)
self.assertIn("Error", result["message"])
@patch("servicenow_mcp.tools.catalog_tools.get_catalog_item_variables")
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_get_catalog_item(self, mock_get, mock_get_variables):
"""Test getting a specific catalog item."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "item1",
"name": "Laptop",
"short_description": "Request a new laptop",
"description": "Request a new laptop for work",
"category": "Hardware",
"price": "1000",
"picture": "laptop.jpg",
"active": "true",
"order": "100",
"delivery_time": "3 days",
"availability": "In Stock",
}
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Mock the variables
mock_get_variables.return_value = [
{
"sys_id": "var1",
"name": "model",
"label": "Laptop Model",
"type": "string",
"mandatory": "true",
"default_value": "MacBook Pro",
"help_text": "Select the laptop model",
"order": "100",
}
]
# Call the function
params = GetCatalogItemParams(item_id="item1")
result = get_catalog_item(self.config, self.auth_manager, params)
# Check the result
self.assertTrue(result.success)
self.assertEqual(result.data["name"], "Laptop")
self.assertEqual(result.data["category"], "Hardware")
self.assertEqual(len(result.data["variables"]), 1)
self.assertEqual(result.data["variables"][0]["name"], "model")
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item/item1")
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_get_catalog_item_not_found(self, mock_get):
"""Test getting a catalog item that doesn't exist."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {"result": {}}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = GetCatalogItemParams(item_id="nonexistent")
result = get_catalog_item(self.config, self.auth_manager, params)
# Check the result
self.assertFalse(result.success)
self.assertIn("not found", result.message)
self.assertIsNone(result.data)
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_get_catalog_item_error(self, mock_get):
"""Test getting a catalog item with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = requests.exceptions.RequestException("Error")
# Call the function
params = GetCatalogItemParams(item_id="item1")
result = get_catalog_item(self.config, self.auth_manager, params)
# Check the result
self.assertFalse(result.success)
self.assertIn("Error", result.message)
self.assertIsNone(result.data)
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_get_catalog_item_variables(self, mock_get):
"""Test getting variables for a catalog item."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "var1",
"name": "model",
"question_text": "Laptop Model",
"type": "string",
"mandatory": "true",
"default_value": "MacBook Pro",
"help_text": "Select the laptop model",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
result = get_catalog_item_variables(self.config, self.auth_manager, "item1")
# Check the result
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["name"], "model")
self.assertEqual(result[0]["label"], "Laptop Model")
self.assertEqual(result[0]["type"], "string")
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/item_option_new")
self.assertEqual(kwargs["params"]["sysparm_query"], "cat_item=item1^ORDERBYorder")
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_get_catalog_item_variables_error(self, mock_get):
"""Test getting variables for a catalog item with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = requests.exceptions.RequestException("Error")
# Call the function
result = get_catalog_item_variables(self.config, self.auth_manager, "item1")
# Check the result
self.assertEqual(len(result), 0)
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_list_catalog_categories(self, mock_get):
"""Test listing catalog categories."""
# Mock the response from ServiceNow
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "cat1",
"title": "Hardware",
"description": "Hardware requests",
"parent": "",
"icon": "hardware.png",
"active": "true",
"order": "100",
}
]
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Call the function
params = ListCatalogCategoriesParams(
limit=10,
offset=0,
query="hardware",
active=True,
)
result = list_catalog_categories(self.config, self.auth_manager, params)
# Check the result
self.assertTrue(result["success"])
self.assertEqual(len(result["categories"]), 1)
self.assertEqual(result["categories"][0]["title"], "Hardware")
self.assertEqual(result["categories"][0]["description"], "Hardware requests")
# Check that the correct URL and parameters were used
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_category")
self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
self.assertIn("sysparm_query", kwargs["params"])
self.assertIn("active=true", kwargs["params"]["sysparm_query"])
self.assertIn("titleLIKEhardware^ORdescriptionLIKEhardware", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.tools.catalog_tools.requests.get")
def test_list_catalog_categories_error(self, mock_get):
"""Test listing catalog categories with an error."""
# Mock the response from ServiceNow
mock_get.side_effect = requests.exceptions.RequestException("Error")
# Call the function
params = ListCatalogCategoriesParams(
limit=10,
offset=0,
)
result = list_catalog_categories(self.config, self.auth_manager, params)
# Check the result
self.assertFalse(result["success"])
self.assertEqual(len(result["categories"]), 0)
self.assertIn("Error", result["message"])
@patch("requests.post")
def test_create_catalog_category(self, mock_post):
"""Test creating a catalog category."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "test_sys_id",
"title": "Test Category",
"description": "Test Description",
"parent": "",
"icon": "icon-test",
"active": "true",
"order": "100",
}
}
mock_post.return_value = mock_response
# Create params
params = CreateCatalogCategoryParams(
title="Test Category",
description="Test Description",
icon="icon-test",
active=True,
order=100,
)
# Call function
result = create_catalog_category(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.data["title"], "Test Category")
self.assertEqual(result.data["sys_id"], "test_sys_id")
# Verify request
mock_post.assert_called_once()
args, kwargs = mock_post.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_category")
self.assertEqual(kwargs["json"]["title"], "Test Category")
self.assertEqual(kwargs["json"]["description"], "Test Description")
@patch("requests.patch")
def test_update_catalog_category(self, mock_patch):
"""Test updating a catalog category."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "test_sys_id",
"title": "Updated Category",
"description": "Updated Description",
"parent": "",
"icon": "icon-test",
"active": "true",
"order": "200",
}
}
mock_patch.return_value = mock_response
# Create params
params = UpdateCatalogCategoryParams(
category_id="test_sys_id",
title="Updated Category",
description="Updated Description",
order=200,
)
# Call function
result = update_catalog_category(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.data["title"], "Updated Category")
self.assertEqual(result.data["description"], "Updated Description")
self.assertEqual(result.data["order"], "200")
# Verify request
mock_patch.assert_called_once()
args, kwargs = mock_patch.call_args
self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_category/test_sys_id")
self.assertEqual(kwargs["json"]["title"], "Updated Category")
self.assertEqual(kwargs["json"]["description"], "Updated Description")
self.assertEqual(kwargs["json"]["order"], "200")
@patch("requests.patch")
def test_move_catalog_items(self, mock_patch):
"""Test moving catalog items."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {"result": {"sys_id": "item_id", "category": "target_category_id"}}
mock_patch.return_value = mock_response
# Create params
params = MoveCatalogItemsParams(
item_ids=["item1", "item2", "item3"],
target_category_id="target_category_id",
)
# Call function
result = move_catalog_items(self.config, self.auth_manager, params)
# Verify result
self.assertTrue(result.success)
self.assertEqual(result.data["moved_items_count"], 3)
# Verify request
self.assertEqual(mock_patch.call_count, 3)
for i, call in enumerate(mock_patch.call_args_list):
args, kwargs = call
self.assertEqual(
args[0],
f"https://example.service-now.com/api/now/table/sc_cat_item/{params.item_ids[i]}"
)
self.assertEqual(kwargs["json"]["category"], "target_category_id")
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/script_include_tools.py:
--------------------------------------------------------------------------------
```python
"""
Script Include tools for the ServiceNow MCP server.
This module provides tools for managing script includes 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 ListScriptIncludesParams(BaseModel):
"""Parameters for listing script includes."""
limit: int = Field(10, description="Maximum number of script includes to return")
offset: int = Field(0, description="Offset for pagination")
active: Optional[bool] = Field(None, description="Filter by active status")
client_callable: Optional[bool] = Field(None, description="Filter by client callable status")
query: Optional[str] = Field(None, description="Search query for script includes")
class GetScriptIncludeParams(BaseModel):
"""Parameters for getting a script include."""
script_include_id: str = Field(..., description="Script include ID or name")
class CreateScriptIncludeParams(BaseModel):
"""Parameters for creating a script include."""
name: str = Field(..., description="Name of the script include")
script: str = Field(..., description="Script content")
description: Optional[str] = Field(None, description="Description of the script include")
api_name: Optional[str] = Field(None, description="API name of the script include")
client_callable: bool = Field(False, description="Whether the script include is client callable")
active: bool = Field(True, description="Whether the script include is active")
access: str = Field("package_private", description="Access level of the script include")
class UpdateScriptIncludeParams(BaseModel):
"""Parameters for updating a script include."""
script_include_id: str = Field(..., description="Script include ID or name")
script: Optional[str] = Field(None, description="Script content")
description: Optional[str] = Field(None, description="Description of the script include")
api_name: Optional[str] = Field(None, description="API name of the script include")
client_callable: Optional[bool] = Field(None, description="Whether the script include is client callable")
active: Optional[bool] = Field(None, description="Whether the script include is active")
access: Optional[str] = Field(None, description="Access level of the script include")
class DeleteScriptIncludeParams(BaseModel):
"""Parameters for deleting a script include."""
script_include_id: str = Field(..., description="Script include ID or name")
class ScriptIncludeResponse(BaseModel):
"""Response from script include operations."""
success: bool = Field(..., description="Whether the operation was successful")
message: str = Field(..., description="Message describing the result")
script_include_id: Optional[str] = Field(None, description="ID of the affected script include")
script_include_name: Optional[str] = Field(None, description="Name of the affected script include")
def list_script_includes(
config: ServerConfig,
auth_manager: AuthManager,
params: ListScriptIncludesParams,
) -> Dict[str, Any]:
"""List script includes from ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for the request.
Returns:
A dictionary containing the list of script includes.
"""
try:
# Build the URL
url = f"{config.instance_url}/api/now/table/sys_script_include"
# Build query parameters
query_params = {
"sysparm_limit": params.limit,
"sysparm_offset": params.offset,
"sysparm_display_value": "true",
"sysparm_exclude_reference_link": "true",
"sysparm_fields": "sys_id,name,script,description,api_name,client_callable,active,access,sys_created_on,sys_updated_on,sys_created_by,sys_updated_by"
}
# Add filters if provided
query_parts = []
if params.active is not None:
query_parts.append(f"active={str(params.active).lower()}")
if params.client_callable is not None:
query_parts.append(f"client_callable={str(params.client_callable).lower()}")
if params.query:
query_parts.append(f"nameLIKE{params.query}")
if query_parts:
query_params["sysparm_query"] = "^".join(query_parts)
# Make the request
headers = auth_manager.get_headers()
response = requests.get(
url,
params=query_params,
headers=headers,
timeout=30,
)
response.raise_for_status()
# Parse the response
data = response.json()
script_includes = []
for item in data.get("result", []):
script_include = {
"sys_id": item.get("sys_id"),
"name": item.get("name"),
"description": item.get("description"),
"api_name": item.get("api_name"),
"client_callable": item.get("client_callable") == "true",
"active": item.get("active") == "true",
"access": item.get("access"),
"created_on": item.get("sys_created_on"),
"updated_on": item.get("sys_updated_on"),
"created_by": item.get("sys_created_by", {}).get("display_value"),
"updated_by": item.get("sys_updated_by", {}).get("display_value"),
}
script_includes.append(script_include)
return {
"success": True,
"message": f"Found {len(script_includes)} script includes",
"script_includes": script_includes,
"total": len(script_includes),
"limit": params.limit,
"offset": params.offset,
}
except Exception as e:
logger.error(f"Error listing script includes: {e}")
return {
"success": False,
"message": f"Error listing script includes: {str(e)}",
"script_includes": [],
"total": 0,
"limit": params.limit,
"offset": params.offset,
}
def get_script_include(
config: ServerConfig,
auth_manager: AuthManager,
params: GetScriptIncludeParams,
) -> Dict[str, Any]:
"""Get a specific script include from ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for the request.
Returns:
A dictionary containing the script include data.
"""
try:
# Build query parameters
query_params = {
"sysparm_display_value": "true",
"sysparm_exclude_reference_link": "true",
"sysparm_fields": "sys_id,name,script,description,api_name,client_callable,active,access,sys_created_on,sys_updated_on,sys_created_by,sys_updated_by"
}
# Determine if we're querying by sys_id or name
if params.script_include_id.startswith("sys_id:"):
sys_id = params.script_include_id.replace("sys_id:", "")
url = f"{config.instance_url}/api/now/table/sys_script_include/{sys_id}"
else:
# Query by name
url = f"{config.instance_url}/api/now/table/sys_script_include"
query_params["sysparm_query"] = f"name={params.script_include_id}"
# Make the request
headers = auth_manager.get_headers()
response = requests.get(
url,
params=query_params,
headers=headers,
timeout=30,
)
response.raise_for_status()
# Parse the response
data = response.json()
if "result" not in data:
return {
"success": False,
"message": f"Script include not found: {params.script_include_id}",
}
# Handle both single result and list of results
result = data["result"]
if isinstance(result, list):
if not result:
return {
"success": False,
"message": f"Script include not found: {params.script_include_id}",
}
item = result[0]
else:
item = result
script_include = {
"sys_id": item.get("sys_id"),
"name": item.get("name"),
"script": item.get("script"),
"description": item.get("description"),
"api_name": item.get("api_name"),
"client_callable": item.get("client_callable") == "true",
"active": item.get("active") == "true",
"access": item.get("access"),
"created_on": item.get("sys_created_on"),
"updated_on": item.get("sys_updated_on"),
"created_by": item.get("sys_created_by", {}).get("display_value"),
"updated_by": item.get("sys_updated_by", {}).get("display_value"),
}
return {
"success": True,
"message": f"Found script include: {item.get('name')}",
"script_include": script_include,
}
except Exception as e:
logger.error(f"Error getting script include: {e}")
return {
"success": False,
"message": f"Error getting script include: {str(e)}",
}
def create_script_include(
config: ServerConfig,
auth_manager: AuthManager,
params: CreateScriptIncludeParams,
) -> ScriptIncludeResponse:
"""Create a new script include in ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for the request.
Returns:
A response indicating the result of the operation.
"""
# Build the URL
url = f"{config.instance_url}/api/now/table/sys_script_include"
# Build the request body
body = {
"name": params.name,
"script": params.script,
"active": str(params.active).lower(),
"client_callable": str(params.client_callable).lower(),
"access": params.access,
}
if params.description:
body["description"] = params.description
if params.api_name:
body["api_name"] = params.api_name
# Make the request
headers = auth_manager.get_headers()
try:
response = requests.post(
url,
json=body,
headers=headers,
timeout=30,
)
response.raise_for_status()
# Parse the response
data = response.json()
if "result" not in data:
return ScriptIncludeResponse(
success=False,
message="Failed to create script include",
)
result = data["result"]
return ScriptIncludeResponse(
success=True,
message=f"Created script include: {result.get('name')}",
script_include_id=result.get("sys_id"),
script_include_name=result.get("name"),
)
except Exception as e:
logger.error(f"Error creating script include: {e}")
return ScriptIncludeResponse(
success=False,
message=f"Error creating script include: {str(e)}",
)
def update_script_include(
config: ServerConfig,
auth_manager: AuthManager,
params: UpdateScriptIncludeParams,
) -> ScriptIncludeResponse:
"""Update an existing script include in ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for the request.
Returns:
A response indicating the result of the operation.
"""
# First, get the script include to update
get_params = GetScriptIncludeParams(script_include_id=params.script_include_id)
get_result = get_script_include(config, auth_manager, get_params)
if not get_result["success"]:
return ScriptIncludeResponse(
success=False,
message=get_result["message"],
)
script_include = get_result["script_include"]
sys_id = script_include["sys_id"]
# Build the URL
url = f"{config.instance_url}/api/now/table/sys_script_include/{sys_id}"
# Build the request body
body = {}
if params.script is not None:
body["script"] = params.script
if params.description is not None:
body["description"] = params.description
if params.api_name is not None:
body["api_name"] = params.api_name
if params.client_callable is not None:
body["client_callable"] = str(params.client_callable).lower()
if params.active is not None:
body["active"] = str(params.active).lower()
if params.access is not None:
body["access"] = params.access
# If no fields to update, return success
if not body:
return ScriptIncludeResponse(
success=True,
message=f"No changes to update for script include: {script_include['name']}",
script_include_id=sys_id,
script_include_name=script_include["name"],
)
# Make the request
headers = auth_manager.get_headers()
try:
response = requests.patch(
url,
json=body,
headers=headers,
timeout=30,
)
response.raise_for_status()
# Parse the response
data = response.json()
if "result" not in data:
return ScriptIncludeResponse(
success=False,
message=f"Failed to update script include: {script_include['name']}",
)
result = data["result"]
return ScriptIncludeResponse(
success=True,
message=f"Updated script include: {result.get('name')}",
script_include_id=result.get("sys_id"),
script_include_name=result.get("name"),
)
except Exception as e:
logger.error(f"Error updating script include: {e}")
return ScriptIncludeResponse(
success=False,
message=f"Error updating script include: {str(e)}",
)
def delete_script_include(
config: ServerConfig,
auth_manager: AuthManager,
params: DeleteScriptIncludeParams,
) -> ScriptIncludeResponse:
"""Delete a script include from ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for the request.
Returns:
A response indicating the result of the operation.
"""
# First, get the script include to delete
get_params = GetScriptIncludeParams(script_include_id=params.script_include_id)
get_result = get_script_include(config, auth_manager, get_params)
if not get_result["success"]:
return ScriptIncludeResponse(
success=False,
message=get_result["message"],
)
script_include = get_result["script_include"]
sys_id = script_include["sys_id"]
name = script_include["name"]
# Build the URL
url = f"{config.instance_url}/api/now/table/sys_script_include/{sys_id}"
# Make the request
headers = auth_manager.get_headers()
try:
response = requests.delete(
url,
headers=headers,
timeout=30,
)
response.raise_for_status()
return ScriptIncludeResponse(
success=True,
message=f"Deleted script include: {name}",
script_include_id=sys_id,
script_include_name=name,
)
except Exception as e:
logger.error(f"Error deleting script include: {e}")
return ScriptIncludeResponse(
success=False,
message=f"Error deleting script include: {str(e)}",
)
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/project_tools.py:
--------------------------------------------------------------------------------
```python
"""
Project management tools for the ServiceNow MCP server.
This module provides tools for managing projects 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 CreateProjectParams(BaseModel):
"""Parameters for creating a project."""
short_description: str = Field(..., description="Project name of the project")
description: Optional[str] = Field(None, description="Detailed description of the project")
status: Optional[str] = Field(None, description="Status of the project (green, yellow, red)")
state: Optional[str] = Field(None, description="State of project (-5 is Pending,1 is Open, 2 is Work in progress, 3 is Closed Complete, 4 is Closed Incomplete, 5 is Closed Skipped)")
project_manager: Optional[str] = Field(None, description="Project manager for the project")
percentage_complete: Optional[int] = Field(None, description="Percentage complete for the project")
assignment_group: Optional[str] = Field(None, description="Group assigned to the project")
assigned_to: Optional[str] = Field(None, description="User assigned to the project")
start_date: Optional[str] = Field(None, description="Start date for the project")
end_date: Optional[str] = Field(None, description="End date for the project")
class UpdateProjectParams(BaseModel):
"""Parameters for updating a project."""
project_id: str = Field(..., description="Project ID or sys_id")
short_description: Optional[str] = Field(None, description="Project name of the project")
description: Optional[str] = Field(None, description="Detailed description of the project")
status: Optional[str] = Field(None, description="Status of the project (green, yellow, red)")
state: Optional[str] = Field(None, description="State of project (-5 is Pending,1 is Open, 2 is Work in progress, 3 is Closed Complete, 4 is Closed Incomplete, 5 is Closed Skipped)")
project_manager: Optional[str] = Field(None, description="Project manager for the project")
percentage_complete: Optional[int] = Field(None, description="Percentage complete for the project")
assignment_group: Optional[str] = Field(None, description="Group assigned to the project")
assigned_to: Optional[str] = Field(None, description="User assigned to the project")
start_date: Optional[str] = Field(None, description="Start date for the project")
end_date: Optional[str] = Field(None, description="End date for the project")
class ListProjectsParams(BaseModel):
"""Parameters for listing projects."""
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")
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")
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_project(
config: ServerConfig, # Changed from auth_manager
auth_manager: AuthManager, # Changed from server_config
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Create a new project in ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for creating the project.
Returns:
The created project.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
CreateProjectParams,
required_fields=["short_description"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {
"short_description": validated_params.short_description,
}
# Add optional fields if provided
if validated_params.description:
data["description"] = validated_params.description
if validated_params.status:
data["status"] = validated_params.status
if validated_params.state:
data["state"] = validated_params.state
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.percentage_complete:
data["percentage_complete"] = validated_params.percentage_complete
if validated_params.assigned_to:
data["assigned_to"] = validated_params.assigned_to
if validated_params.project_manager:
data["project_manager"] = validated_params.project_manager
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, 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, 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/pm_project"
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Project created successfully",
"project": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error creating project: {e}")
return {
"success": False,
"message": f"Error creating project: {str(e)}",
}
def update_project(
config: ServerConfig, # Changed from auth_manager
auth_manager: AuthManager, # Changed from server_config
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
Update an existing project in ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for updating the project.
Returns:
The updated project.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
UpdateProjectParams,
required_fields=["project_id"]
)
if not result["success"]:
return result
validated_params = result["params"]
# Prepare the request data
data = {}
# Add optional 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.status:
data["status"] = validated_params.status
if validated_params.state:
data["state"] = validated_params.state
if validated_params.assignment_group:
data["assignment_group"] = validated_params.assignment_group
if validated_params.percentage_complete:
data["percentage_complete"] = validated_params.percentage_complete
if validated_params.assigned_to:
data["assigned_to"] = validated_params.assigned_to
if validated_params.project_manager:
data["project_manager"] = validated_params.project_manager
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, 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, 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/pm_project/{validated_params.project_id}"
try:
response = requests.put(url, json=data, headers=headers)
response.raise_for_status()
result = response.json()
return {
"success": True,
"message": "Project updated successfully",
"project": result["result"],
}
except requests.exceptions.RequestException as e:
logger.error(f"Error updating project: {e}")
return {
"success": False,
"message": f"Error updating project: {str(e)}",
}
def list_projects(
config: ServerConfig, # Changed from auth_manager
auth_manager: AuthManager, # Changed from server_config
params: Dict[str, Any],
) -> Dict[str, Any]:
"""
List projects from ServiceNow.
Args:
config: The server configuration.
auth_manager: The authentication manager.
params: The parameters for listing projects.
Returns:
A list of projects.
"""
# Unwrap and validate parameters
result = _unwrap_and_validate_params(
params,
ListProjectsParams
)
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.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, 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, 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/pm_project"
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
projects = result.get("result", [])
count = len(projects)
return {
"success": True,
"projects": projects,
"count": count,
"total": count, # Use count as total if total is not provided
}
except requests.exceptions.RequestException as e:
logger.error(f"Error listing projects: {e}")
return {
"success": False,
"message": f"Error listing projects: {str(e)}",
}
```
--------------------------------------------------------------------------------
/tests/test_script_include_tools.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the script include tools.
This module contains tests for the script include tools in the ServiceNow MCP server.
"""
import unittest
import requests
from unittest.mock import MagicMock, patch
from servicenow_mcp.auth.auth_manager import AuthManager
from servicenow_mcp.tools.script_include_tools import (
ListScriptIncludesParams,
GetScriptIncludeParams,
CreateScriptIncludeParams,
UpdateScriptIncludeParams,
DeleteScriptIncludeParams,
ScriptIncludeResponse,
list_script_includes,
get_script_include,
create_script_include,
update_script_include,
delete_script_include,
)
from servicenow_mcp.utils.config import ServerConfig, AuthConfig, AuthType, BasicAuthConfig
class TestScriptIncludeTools(unittest.TestCase):
"""Tests for the script include tools."""
def setUp(self):
"""Set up test fixtures."""
auth_config = AuthConfig(
type=AuthType.BASIC,
basic=BasicAuthConfig(
username="test_user",
password="test_password"
)
)
self.server_config = ServerConfig(
instance_url="https://test.service-now.com",
auth=auth_config,
)
self.auth_manager = MagicMock(spec=AuthManager)
self.auth_manager.get_headers.return_value = {
"Authorization": "Bearer test",
"Content-Type": "application/json",
}
@patch("servicenow_mcp.tools.script_include_tools.requests.get")
def test_list_script_includes(self, mock_get):
"""Test listing script includes."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": [
{
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": "true",
"active": "true",
"access": "public",
"sys_created_on": "2023-01-01 00:00:00",
"sys_updated_on": "2023-01-02 00:00:00",
"sys_created_by": {"display_value": "admin"},
"sys_updated_by": {"display_value": "admin"}
}
]
}
mock_response.status_code = 200
mock_get.return_value = mock_response
# Call the method
params = ListScriptIncludesParams(
limit=10,
offset=0,
active=True,
client_callable=True,
query="Test"
)
result = list_script_includes(self.server_config, self.auth_manager, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual(1, len(result["script_includes"]))
self.assertEqual("123", result["script_includes"][0]["sys_id"])
self.assertEqual("TestScriptInclude", result["script_includes"][0]["name"])
self.assertTrue(result["script_includes"][0]["client_callable"])
self.assertTrue(result["script_includes"][0]["active"])
# Verify the request
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
self.assertEqual(10, kwargs["params"]["sysparm_limit"])
self.assertEqual(0, kwargs["params"]["sysparm_offset"])
self.assertEqual("active=true^client_callable=true^nameLIKETest", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.tools.script_include_tools.requests.get")
def test_get_script_include(self, mock_get):
"""Test getting a script include."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": "true",
"active": "true",
"access": "public",
"sys_created_on": "2023-01-01 00:00:00",
"sys_updated_on": "2023-01-02 00:00:00",
"sys_created_by": {"display_value": "admin"},
"sys_updated_by": {"display_value": "admin"}
}
}
mock_response.status_code = 200
mock_get.return_value = mock_response
# Call the method
params = GetScriptIncludeParams(script_include_id="123")
result = get_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertTrue(result["success"])
self.assertEqual("123", result["script_include"]["sys_id"])
self.assertEqual("TestScriptInclude", result["script_include"]["name"])
self.assertTrue(result["script_include"]["client_callable"])
self.assertTrue(result["script_include"]["active"])
# Verify the request
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
self.assertEqual("name=123", kwargs["params"]["sysparm_query"])
@patch("servicenow_mcp.tools.script_include_tools.requests.post")
def test_create_script_include(self, mock_post):
"""Test creating a script include."""
# Mock response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "123",
"name": "TestScriptInclude",
}
}
mock_response.status_code = 201
mock_post.return_value = mock_response
# Call the method
params = CreateScriptIncludeParams(
name="TestScriptInclude",
script="var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
description="Test Script Include",
api_name="global.TestScriptInclude",
client_callable=True,
active=True,
access="public"
)
result = create_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertTrue(result.success)
self.assertEqual("123", result.script_include_id)
self.assertEqual("TestScriptInclude", result.script_include_name)
# Verify the request
mock_post.assert_called_once()
args, kwargs = mock_post.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
self.assertEqual("TestScriptInclude", kwargs["json"]["name"])
self.assertEqual("true", kwargs["json"]["client_callable"])
self.assertEqual("true", kwargs["json"]["active"])
self.assertEqual("public", kwargs["json"]["access"])
@patch("servicenow_mcp.tools.script_include_tools.get_script_include")
@patch("servicenow_mcp.tools.script_include_tools.requests.patch")
def test_update_script_include(self, mock_patch, mock_get_script_include):
"""Test updating a script include."""
# Mock get_script_include response
mock_get_script_include.return_value = {
"success": True,
"message": "Found script include: TestScriptInclude",
"script_include": {
"sys_id": "123",
"name": "TestScriptInclude",
"script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
"description": "Test Script Include",
"api_name": "global.TestScriptInclude",
"client_callable": True,
"active": True,
"access": "public",
}
}
# Mock patch response
mock_response = MagicMock()
mock_response.json.return_value = {
"result": {
"sys_id": "123",
"name": "TestScriptInclude",
}
}
mock_response.status_code = 200
mock_patch.return_value = mock_response
# Call the method
params = UpdateScriptIncludeParams(
script_include_id="123",
script="var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n // Updated\n },\n\n type: 'TestScriptInclude'\n};",
description="Updated Test Script Include",
client_callable=False,
)
result = update_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertTrue(result.success)
self.assertEqual("123", result.script_include_id)
self.assertEqual("TestScriptInclude", result.script_include_name)
# Verify the request
mock_patch.assert_called_once()
args, kwargs = mock_patch.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include/123", args[0])
self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
self.assertEqual("Updated Test Script Include", kwargs["json"]["description"])
self.assertEqual("false", kwargs["json"]["client_callable"])
@patch("servicenow_mcp.tools.script_include_tools.get_script_include")
@patch("servicenow_mcp.tools.script_include_tools.requests.delete")
def test_delete_script_include(self, mock_delete, mock_get_script_include):
"""Test deleting a script include."""
# Mock get_script_include response
mock_get_script_include.return_value = {
"success": True,
"message": "Found script include: TestScriptInclude",
"script_include": {
"sys_id": "123",
"name": "TestScriptInclude",
}
}
# Mock delete response
mock_response = MagicMock()
mock_response.status_code = 204
mock_delete.return_value = mock_response
# Call the method
params = DeleteScriptIncludeParams(script_include_id="123")
result = delete_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertTrue(result.success)
self.assertEqual("123", result.script_include_id)
self.assertEqual("TestScriptInclude", result.script_include_name)
# Verify the request
mock_delete.assert_called_once()
args, kwargs = mock_delete.call_args
self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include/123", args[0])
self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
@patch("servicenow_mcp.tools.script_include_tools.requests.get")
def test_list_script_includes_error(self, mock_get):
"""Test listing script includes with an error."""
# Mock response
mock_get.side_effect = requests.RequestException("Test error")
# Call the method
params = ListScriptIncludesParams()
result = list_script_includes(self.server_config, self.auth_manager, params)
# Verify the result
self.assertFalse(result["success"])
self.assertIn("Error listing script includes", result["message"])
@patch("servicenow_mcp.tools.script_include_tools.requests.get")
def test_get_script_include_error(self, mock_get):
"""Test getting a script include with an error."""
# Mock response
mock_get.side_effect = requests.RequestException("Test error")
# Call the method
params = GetScriptIncludeParams(script_include_id="123")
result = get_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertFalse(result["success"])
self.assertIn("Error getting script include", result["message"])
@patch("servicenow_mcp.tools.script_include_tools.requests.post")
def test_create_script_include_error(self, mock_post):
"""Test creating a script include with an error."""
# Mock response
mock_post.side_effect = requests.RequestException("Test error")
# Call the method
params = CreateScriptIncludeParams(
name="TestScriptInclude",
script="var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
)
result = create_script_include(self.server_config, self.auth_manager, params)
# Verify the result
self.assertFalse(result.success)
self.assertIn("Error creating script include", result.message)
class TestScriptIncludeParams(unittest.TestCase):
"""Tests for the script include parameters."""
def test_list_script_includes_params(self):
"""Test list script includes parameters."""
params = ListScriptIncludesParams(
limit=20,
offset=10,
active=True,
client_callable=False,
query="Test"
)
self.assertEqual(20, params.limit)
self.assertEqual(10, params.offset)
self.assertTrue(params.active)
self.assertFalse(params.client_callable)
self.assertEqual("Test", params.query)
def test_get_script_include_params(self):
"""Test get script include parameters."""
params = GetScriptIncludeParams(script_include_id="123")
self.assertEqual("123", params.script_include_id)
def test_create_script_include_params(self):
"""Test create script include parameters."""
params = CreateScriptIncludeParams(
name="TestScriptInclude",
script="var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
description="Test Script Include",
api_name="global.TestScriptInclude",
client_callable=True,
active=True,
access="public"
)
self.assertEqual("TestScriptInclude", params.name)
self.assertTrue(params.client_callable)
self.assertTrue(params.active)
self.assertEqual("public", params.access)
def test_update_script_include_params(self):
"""Test update script include parameters."""
params = UpdateScriptIncludeParams(
script_include_id="123",
script="var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n // Updated\n },\n\n type: 'TestScriptInclude'\n};",
description="Updated Test Script Include",
client_callable=False,
)
self.assertEqual("123", params.script_include_id)
self.assertEqual("Updated Test Script Include", params.description)
self.assertFalse(params.client_callable)
def test_delete_script_include_params(self):
"""Test delete script include parameters."""
params = DeleteScriptIncludeParams(script_include_id="123")
self.assertEqual("123", params.script_include_id)
def test_script_include_response(self):
"""Test script include response."""
response = ScriptIncludeResponse(
success=True,
message="Test message",
script_include_id="123",
script_include_name="TestScriptInclude"
)
self.assertTrue(response.success)
self.assertEqual("Test message", response.message)
self.assertEqual("123", response.script_include_id)
self.assertEqual("TestScriptInclude", response.script_include_name)
```