This is page 2 of 5. Use http://codebase.md/osomai/servicenow-mcp?lines=true&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
--------------------------------------------------------------------------------
/examples/change_management_demo.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """
3 | Change Management Demo
4 |
5 | This script demonstrates how to use the ServiceNow MCP server with Claude Desktop
6 | to manage change requests in ServiceNow.
7 | """
8 |
9 | import json
10 | import os
11 | from datetime import datetime, timedelta
12 |
13 | from dotenv import load_dotenv
14 |
15 | from servicenow_mcp.auth.auth_manager import AuthManager
16 | from servicenow_mcp.tools.change_tools import (
17 | add_change_task,
18 | approve_change,
19 | create_change_request,
20 | get_change_request_details,
21 | list_change_requests,
22 | submit_change_for_approval,
23 | update_change_request,
24 | )
25 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
26 |
27 |
28 | def main():
29 | """Run the change management demo."""
30 | # Load environment variables from .env file
31 | load_dotenv()
32 |
33 | # Create auth configuration
34 | auth_config = AuthConfig(
35 | type=AuthType.BASIC,
36 | basic=BasicAuthConfig(
37 | username=os.environ.get("SERVICENOW_USERNAME"),
38 | password=os.environ.get("SERVICENOW_PASSWORD"),
39 | )
40 | )
41 |
42 | # Create server configuration with auth
43 | server_config = ServerConfig(
44 | instance_url=os.environ.get("SERVICENOW_INSTANCE_URL"),
45 | debug=os.environ.get("DEBUG", "false").lower() == "true",
46 | auth=auth_config,
47 | )
48 |
49 | # Create authentication manager with the auth_config
50 | auth_manager = AuthManager(auth_config)
51 |
52 | print("ServiceNow Change Management Demo")
53 | print("=================================")
54 | print(f"Instance URL: {server_config.instance_url}")
55 | print(f"Auth Type: {auth_config.type}")
56 | print()
57 |
58 | # Demo 1: Create a change request
59 | print("Demo 1: Creating a change request")
60 | print("---------------------------------")
61 |
62 | # Calculate start and end dates for the change (tomorrow)
63 | tomorrow = datetime.now() + timedelta(days=1)
64 | start_date = tomorrow.replace(hour=1, minute=0, second=0).strftime("%Y-%m-%d %H:%M:%S")
65 | end_date = tomorrow.replace(hour=3, minute=0, second=0).strftime("%Y-%m-%d %H:%M:%S")
66 |
67 | create_params = {
68 | "short_description": "Server maintenance - Apply security patches",
69 | "description": "Apply the latest security patches to the application servers to address recent vulnerabilities.",
70 | "type": "normal",
71 | "risk": "moderate",
72 | "impact": "medium",
73 | "category": "Hardware",
74 | "start_date": start_date,
75 | "end_date": end_date,
76 | }
77 |
78 | result = create_change_request(auth_manager, server_config, create_params)
79 | print(json.dumps(result, indent=2))
80 |
81 | if not result.get("success"):
82 | print("Failed to create change request. Exiting demo.")
83 | return
84 |
85 | # Store the change request ID for later use
86 | change_id = result["change_request"]["sys_id"]
87 | print(f"Created change request with ID: {change_id}")
88 | print()
89 |
90 | # Demo 2: Add tasks to the change request
91 | print("Demo 2: Adding tasks to the change request")
92 | print("-----------------------------------------")
93 |
94 | # Task 1: Pre-implementation checks
95 | task1_params = {
96 | "change_id": change_id,
97 | "short_description": "Pre-implementation checks",
98 | "description": "Verify system backups and confirm all prerequisites are met.",
99 | "planned_start_date": start_date,
100 | "planned_end_date": (datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S"),
101 | }
102 |
103 | task1_result = add_change_task(auth_manager, server_config, task1_params)
104 | print(json.dumps(task1_result, indent=2))
105 |
106 | # Task 2: Apply patches
107 | task2_params = {
108 | "change_id": change_id,
109 | "short_description": "Apply security patches",
110 | "description": "Apply the latest security patches to all application servers.",
111 | "planned_start_date": (datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S"),
112 | "planned_end_date": (datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=90)).strftime("%Y-%m-%d %H:%M:%S"),
113 | }
114 |
115 | task2_result = add_change_task(auth_manager, server_config, task2_params)
116 | print(json.dumps(task2_result, indent=2))
117 |
118 | # Task 3: Post-implementation verification
119 | task3_params = {
120 | "change_id": change_id,
121 | "short_description": "Post-implementation verification",
122 | "description": "Verify all systems are functioning correctly after patching.",
123 | "planned_start_date": (datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=90)).strftime("%Y-%m-%d %H:%M:%S"),
124 | "planned_end_date": (datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=120)).strftime("%Y-%m-%d %H:%M:%S"),
125 | }
126 |
127 | task3_result = add_change_task(auth_manager, server_config, task3_params)
128 | print(json.dumps(task3_result, indent=2))
129 | print()
130 |
131 | # Demo 3: Update the change request
132 | print("Demo 3: Updating the change request")
133 | print("----------------------------------")
134 |
135 | update_params = {
136 | "change_id": change_id,
137 | "work_notes": "Added implementation plan and tasks. Ready for review.",
138 | "risk": "low", # Downgrading risk after assessment
139 | }
140 |
141 | update_result = update_change_request(auth_manager, server_config, update_params)
142 | print(json.dumps(update_result, indent=2))
143 | print()
144 |
145 | # Demo 4: Get change request details
146 | print("Demo 4: Getting change request details")
147 | print("-------------------------------------")
148 |
149 | details_params = {
150 | "change_id": change_id,
151 | }
152 |
153 | details_result = get_change_request_details(auth_manager, server_config, details_params)
154 | print(json.dumps(details_result, indent=2))
155 | print()
156 |
157 | # Demo 5: Submit change for approval
158 | print("Demo 5: Submitting change for approval")
159 | print("-------------------------------------")
160 |
161 | approval_params = {
162 | "change_id": change_id,
163 | "approval_comments": "Implementation plan is complete and ready for review.",
164 | }
165 |
166 | approval_result = submit_change_for_approval(auth_manager, server_config, approval_params)
167 | print(json.dumps(approval_result, indent=2))
168 | print()
169 |
170 | # Demo 6: List change requests
171 | print("Demo 6: Listing change requests")
172 | print("------------------------------")
173 |
174 | list_params = {
175 | "limit": 5,
176 | "timeframe": "upcoming",
177 | }
178 |
179 | list_result = list_change_requests(auth_manager, server_config, list_params)
180 | print(json.dumps(list_result, indent=2))
181 | print()
182 |
183 | # Demo 7: Approve the change request
184 | print("Demo 7: Approving the change request")
185 | print("-----------------------------------")
186 |
187 | approve_params = {
188 | "change_id": change_id,
189 | "approval_comments": "Implementation plan looks good. Approved.",
190 | }
191 |
192 | approve_result = approve_change(auth_manager, server_config, approve_params)
193 | print(json.dumps(approve_result, indent=2))
194 | print()
195 |
196 | print("Change Management Demo Complete")
197 | print("==============================")
198 | print(f"Created and managed change request: {change_id}")
199 | print("You can now view this change request in your ServiceNow instance.")
200 |
201 |
202 | if __name__ == "__main__":
203 | main()
```
--------------------------------------------------------------------------------
/docs/change_management.md:
--------------------------------------------------------------------------------
```markdown
1 | # ServiceNow MCP Change Management Tools
2 |
3 | This document provides information about the change management tools available in the ServiceNow MCP server.
4 |
5 | ## Overview
6 |
7 | The change management tools allow Claude to interact with ServiceNow's change management functionality, enabling users to create, update, and manage change requests through natural language conversations.
8 |
9 | ## Available Tools
10 |
11 | The ServiceNow MCP server provides the following change management tools:
12 |
13 | ### Core Change Request Management
14 |
15 | 1. **create_change_request** - Create a new change request in ServiceNow
16 | - Parameters:
17 | - `short_description` (required): Short description of the change request
18 | - `description`: Detailed description of the change request
19 | - `type` (required): Type of change (normal, standard, emergency)
20 | - `risk`: Risk level of the change
21 | - `impact`: Impact of the change
22 | - `category`: Category of the change
23 | - `requested_by`: User who requested the change
24 | - `assignment_group`: Group assigned to the change
25 | - `start_date`: Planned start date (YYYY-MM-DD HH:MM:SS)
26 | - `end_date`: Planned end date (YYYY-MM-DD HH:MM:SS)
27 |
28 | 2. **update_change_request** - Update an existing change request
29 | - Parameters:
30 | - `change_id` (required): Change request ID or sys_id
31 | - `short_description`: Short description of the change request
32 | - `description`: Detailed description of the change request
33 | - `state`: State of the change request
34 | - `risk`: Risk level of the change
35 | - `impact`: Impact of the change
36 | - `category`: Category of the change
37 | - `assignment_group`: Group assigned to the change
38 | - `start_date`: Planned start date (YYYY-MM-DD HH:MM:SS)
39 | - `end_date`: Planned end date (YYYY-MM-DD HH:MM:SS)
40 | - `work_notes`: Work notes to add to the change request
41 |
42 | 3. **list_change_requests** - List change requests with filtering options
43 | - Parameters:
44 | - `limit`: Maximum number of records to return (default: 10)
45 | - `offset`: Offset to start from (default: 0)
46 | - `state`: Filter by state
47 | - `type`: Filter by type (normal, standard, emergency)
48 | - `category`: Filter by category
49 | - `assignment_group`: Filter by assignment group
50 | - `timeframe`: Filter by timeframe (upcoming, in-progress, completed)
51 | - `query`: Additional query string
52 |
53 | 4. **get_change_request_details** - Get detailed information about a specific change request
54 | - Parameters:
55 | - `change_id` (required): Change request ID or sys_id
56 |
57 | 5. **add_change_task** - Add a task to a change request
58 | - Parameters:
59 | - `change_id` (required): Change request ID or sys_id
60 | - `short_description` (required): Short description of the task
61 | - `description`: Detailed description of the task
62 | - `assigned_to`: User assigned to the task
63 | - `planned_start_date`: Planned start date (YYYY-MM-DD HH:MM:SS)
64 | - `planned_end_date`: Planned end date (YYYY-MM-DD HH:MM:SS)
65 |
66 | ### Change Approval Workflow
67 |
68 | 1. **submit_change_for_approval** - Submit a change request for approval
69 | - Parameters:
70 | - `change_id` (required): Change request ID or sys_id
71 | - `approval_comments`: Comments for the approval request
72 |
73 | 2. **approve_change** - Approve a change request
74 | - Parameters:
75 | - `change_id` (required): Change request ID or sys_id
76 | - `approver_id`: ID of the approver
77 | - `approval_comments`: Comments for the approval
78 |
79 | 3. **reject_change** - Reject a change request
80 | - Parameters:
81 | - `change_id` (required): Change request ID or sys_id
82 | - `approver_id`: ID of the approver
83 | - `rejection_reason` (required): Reason for rejection
84 |
85 | ## Example Usage with Claude
86 |
87 | Once the ServiceNow MCP server is configured with Claude Desktop, you can ask Claude to perform actions like:
88 |
89 | ### Creating and Managing Change Requests
90 |
91 | - "Create a change request for server maintenance to apply security patches tomorrow night"
92 | - "Schedule a database upgrade for next Tuesday from 2 AM to 4 AM"
93 | - "Create an emergency change to fix the critical security vulnerability in our web application"
94 |
95 | ### Adding Tasks and Implementation Details
96 |
97 | - "Add a task to the server maintenance change for pre-implementation checks"
98 | - "Add a task to verify system backups before starting the database upgrade"
99 | - "Update the implementation plan for the network change to include rollback procedures"
100 |
101 | ### Approval Workflow
102 |
103 | - "Submit the server maintenance change for approval"
104 | - "Show me all changes waiting for my approval"
105 | - "Approve the database upgrade change with comment: implementation plan looks thorough"
106 | - "Reject the network change due to insufficient testing"
107 |
108 | ### Querying Change Information
109 |
110 | - "Show me all emergency changes scheduled for this week"
111 | - "What's the status of the database upgrade change?"
112 | - "List all changes assigned to the Network team"
113 | - "Show me the details of change CHG0010001"
114 |
115 | ## Example Code
116 |
117 | Here's an example of how to use the change management tools programmatically:
118 |
119 | ```python
120 | from servicenow_mcp.auth.auth_manager import AuthManager
121 | from servicenow_mcp.tools.change_tools import create_change_request
122 | from servicenow_mcp.utils.config import ServerConfig
123 |
124 | # Create server configuration
125 | server_config = ServerConfig(
126 | instance_url="https://your-instance.service-now.com",
127 | )
128 |
129 | # Create authentication manager
130 | auth_manager = AuthManager(
131 | auth_type="basic",
132 | username="your-username",
133 | password="your-password",
134 | instance_url="https://your-instance.service-now.com",
135 | )
136 |
137 | # Create a change request
138 | create_params = {
139 | "short_description": "Server maintenance - Apply security patches",
140 | "description": "Apply the latest security patches to the application servers.",
141 | "type": "normal",
142 | "risk": "moderate",
143 | "impact": "medium",
144 | "category": "Hardware",
145 | "start_date": "2023-12-15 01:00:00",
146 | "end_date": "2023-12-15 03:00:00",
147 | }
148 |
149 | result = create_change_request(auth_manager, server_config, create_params)
150 | print(result)
151 | ```
152 |
153 | For a complete example, see the [change_management_demo.py](../examples/change_management_demo.py) script.
154 |
155 | ## Integration with Claude Desktop
156 |
157 | To configure the ServiceNow MCP server with change management tools in Claude Desktop:
158 |
159 | 1. Edit the Claude Desktop configuration file at `~/Library/Application Support/Claude/claude_desktop_config.json` (macOS) or the appropriate path for your OS:
160 |
161 | ```json
162 | {
163 | "mcpServers": {
164 | "ServiceNow": {
165 | "command": "/Users/yourusername/dev/servicenow-mcp/.venv/bin/python",
166 | "args": [
167 | "-m",
168 | "servicenow_mcp.cli"
169 | ],
170 | "env": {
171 | "SERVICENOW_INSTANCE_URL": "https://your-instance.service-now.com",
172 | "SERVICENOW_USERNAME": "your-username",
173 | "SERVICENOW_PASSWORD": "your-password",
174 | "SERVICENOW_AUTH_TYPE": "basic"
175 | }
176 | }
177 | }
178 | }
179 | ```
180 |
181 | 2. Restart Claude Desktop to apply the changes
182 |
183 | ## Customization
184 |
185 | The change management tools can be customized to match your organization's specific ServiceNow configuration:
186 |
187 | - State values may need to be adjusted based on your ServiceNow instance configuration
188 | - Additional fields can be added to the parameter models if needed
189 | - Approval workflows may need to be modified to match your organization's approval process
190 |
191 | To customize the tools, modify the `change_tools.py` file in the `src/servicenow_mcp/tools` directory.
```
--------------------------------------------------------------------------------
/docs/knowledge_base.md:
--------------------------------------------------------------------------------
```markdown
1 | # Knowledge Base Management
2 |
3 | The ServiceNow MCP server provides tools for managing knowledge bases, categories, and articles in ServiceNow.
4 |
5 | ## Overview
6 |
7 | Knowledge bases in ServiceNow allow organizations to store and share information in a structured format. This can include documentation, FAQs, troubleshooting guides, and other knowledge resources.
8 |
9 | The ServiceNow MCP Knowledge Base tools provide a way to create and manage knowledge bases, categories, and articles through the ServiceNow API.
10 |
11 | ## Available Tools
12 |
13 | ### Knowledge Base Management
14 |
15 | 1. **create_knowledge_base** - Create a new knowledge base in ServiceNow
16 | - Parameters:
17 | - `title` - Title of the knowledge base
18 | - `description` (optional) - Description of the knowledge base
19 | - `owner` (optional) - The specified admin user or group
20 | - `managers` (optional) - Users who can manage this knowledge base
21 | - `publish_workflow` (optional) - Publication workflow
22 | - `retire_workflow` (optional) - Retirement workflow
23 |
24 | 2. **list_knowledge_bases** - List knowledge bases from ServiceNow
25 | - Parameters:
26 | - `limit` (optional, default: 10) - Maximum number of knowledge bases to return
27 | - `offset` (optional, default: 0) - Offset for pagination
28 | - `active` (optional) - Filter by active status
29 | - `query` (optional) - Search query for knowledge bases
30 |
31 | ### Category Management
32 |
33 | 1. **create_category** - Create a new category in a knowledge base
34 | - Parameters:
35 | - `title` - Title of the category
36 | - `description` (optional) - Description of the category
37 | - `knowledge_base` - The knowledge base to create the category in
38 | - `parent_category` (optional) - Parent category (if creating a subcategory)
39 | - `active` (optional, default: true) - Whether the category is active
40 |
41 | 2. **list_categories** - List categories in a knowledge base
42 | - Parameters:
43 | - `knowledge_base` (optional) - Filter by knowledge base ID
44 | - `parent_category` (optional) - Filter by parent category ID
45 | - `limit` (optional, default: 10) - Maximum number of categories to return
46 | - `offset` (optional, default: 0) - Offset for pagination
47 | - `active` (optional) - Filter by active status
48 | - `query` (optional) - Search query for categories
49 |
50 | ### Article Management
51 |
52 | 1. **create_article** - Create a new knowledge article
53 | - Parameters:
54 | - `title` - Title of the article
55 | - `text` - The main body text for the article
56 | - `short_description` - Short description of the article
57 | - `knowledge_base` - The knowledge base to create the article in
58 | - `category` - Category for the article
59 | - `keywords` (optional) - Keywords for search
60 | - `article_type` (optional, default: "text") - The type of article
61 |
62 | 2. **update_article** - Update an existing knowledge article
63 | - Parameters:
64 | - `article_id` - ID of the article to update
65 | - `title` (optional) - Updated title of the article
66 | - `text` (optional) - Updated main body text for the article
67 | - `short_description` (optional) - Updated short description
68 | - `category` (optional) - Updated category for the article
69 | - `keywords` (optional) - Updated keywords for search
70 |
71 | 3. **publish_article** - Publish a knowledge article
72 | - Parameters:
73 | - `article_id` - ID of the article to publish
74 | - `workflow_state` (optional, default: "published") - The workflow state to set
75 | - `workflow_version` (optional) - The workflow version to use
76 |
77 | 4. **list_articles** - List knowledge articles with filtering options
78 | - Parameters:
79 | - `limit` (optional, default: 10) - Maximum number of articles to return
80 | - `offset` (optional, default: 0) - Offset for pagination
81 | - `knowledge_base` (optional) - Filter by knowledge base
82 | - `category` (optional) - Filter by category
83 | - `query` (optional) - Search query for articles
84 | - `workflow_state` (optional) - Filter by workflow state
85 |
86 | 5. **get_article** - Get a specific knowledge article by ID
87 | - Parameters:
88 | - `article_id` - ID of the article to get
89 |
90 | ## Example Usage
91 |
92 | ### Creating a Knowledge Base
93 |
94 | ```python
95 | response = create_knowledge_base({
96 | "title": "Healthcare IT Knowledge Base",
97 | "description": "Knowledge base for healthcare IT resources and documentation",
98 | "owner": "healthcare_admins"
99 | })
100 | print(f"Knowledge base created with ID: {response['kb_id']}")
101 | ```
102 |
103 | ### Listing Knowledge Bases
104 |
105 | ```python
106 | response = list_knowledge_bases({
107 | "active": True,
108 | "query": "IT",
109 | "limit": 20
110 | })
111 | print(f"Found {response['count']} knowledge bases")
112 | for kb in response['knowledge_bases']:
113 | print(f"- {kb['title']}")
114 | ```
115 |
116 | ### Creating a Category
117 |
118 | ```python
119 | response = create_category({
120 | "title": "Network Configuration",
121 | "description": "Articles related to network configuration in healthcare environments",
122 | "knowledge_base": "healthcare_kb"
123 | })
124 | print(f"Category created with ID: {response['category_id']}")
125 | ```
126 |
127 | ### Creating an Article
128 |
129 | ```python
130 | response = create_article({
131 | "title": "VPN Setup for Remote Clinicians",
132 | "short_description": "Step-by-step guide for setting up VPN access for remote clinicians",
133 | "text": "# VPN Setup Guide\n\n1. Install the VPN client...",
134 | "knowledge_base": "healthcare_kb",
135 | "category": "network_config",
136 | "keywords": "vpn, remote access, clinicians, setup"
137 | })
138 | print(f"Article created with ID: {response['article_id']}")
139 | ```
140 |
141 | ### Updating an Article
142 |
143 | ```python
144 | response = update_article({
145 | "article_id": "kb0010001",
146 | "text": "# Updated VPN Setup Guide\n\n1. Download the latest VPN client...",
147 | "keywords": "vpn, remote access, clinicians, setup, configuration"
148 | })
149 | print(f"Article updated: {response['success']}")
150 | ```
151 |
152 | ### Publishing an Article
153 |
154 | ```python
155 | response = publish_article({
156 | "article_id": "kb0010001"
157 | })
158 | print(f"Article published: {response['success']}")
159 | ```
160 |
161 | ### Listing Articles
162 |
163 | ```python
164 | response = list_articles({
165 | "knowledge_base": "healthcare_kb",
166 | "category": "network_config",
167 | "limit": 20
168 | })
169 | print(f"Found {response['count']} articles")
170 | for article in response['articles']:
171 | print(f"- {article['title']}")
172 | ```
173 |
174 | ### Getting Article Details
175 |
176 | ```python
177 | response = get_article({
178 | "article_id": "kb0010001"
179 | })
180 | article = response['article']
181 | print(f"Title: {article['title']}")
182 | print(f"Category: {article['category']}")
183 | print(f"Views: {article['views']}")
184 | ```
185 |
186 | ### Listing Categories
187 |
188 | ```python
189 | response = list_categories({
190 | "knowledge_base": "healthcare_kb",
191 | "active": True,
192 | "limit": 20
193 | })
194 | print(f"Found {response['count']} categories")
195 | for category in response['categories']:
196 | print(f"- {category['title']}")
197 | ```
198 |
199 | ## ServiceNow API Endpoints
200 |
201 | The Knowledge Base tools use the following ServiceNow API endpoints:
202 |
203 | - `/table/kb_knowledge_base` - Knowledge base table API
204 | - `/table/kb_category` - Category table API
205 | - `/table/kb_knowledge` - Knowledge article table API
206 |
207 | ## Error Handling
208 |
209 | All knowledge base tools handle errors gracefully and return responses with `success` and `message` fields. If an operation fails, the `success` field will be `false` and the `message` field will contain information about the error.
210 |
211 | ## Additional Information
212 |
213 | For more information about knowledge management in ServiceNow, see the [ServiceNow Knowledge Management documentation](https://docs.servicenow.com/bundle/tokyo-servicenow-platform/page/product/knowledge-management/concept/c_KnowledgeManagement.html).
```
--------------------------------------------------------------------------------
/prompts/add_servicenow_mcp_tool.md:
--------------------------------------------------------------------------------
```markdown
1 | # ServiceNow MCP Tool Creation Prompt
2 |
3 | Use this prompt to guide the development of new tools for the ServiceNow MCP project. This structured approach ensures consistency in implementation, testing, and documentation.
4 |
5 | ## Background
6 |
7 | The ServiceNow MCP (Model Completion Protocol) server allows Claude to interact with ServiceNow instances, retrieving data and performing actions through the ServiceNow API. Adding new tools expands the capabilities of this bridge.
8 |
9 | ## Required Files to Create/Modify
10 |
11 | For each new tool, you need to:
12 |
13 | 1. Create/modify a tool module in `src/servicenow_mcp/tools/`
14 | 2. Update the tools `__init__.py` to expose the new tool
15 | 3. Update `server.py` to register the tool with the MCP server
16 | 4. Add unit tests in the `tests/` directory
17 | 5. Update documentation in the `docs/` directory
18 | 6. Update the `README.md` to include the new tool
19 |
20 | ## Implementation Steps
21 |
22 | Please implement the following ServiceNow tool capability: {DESCRIBE_CAPABILITY_HERE}
23 |
24 | Follow these steps to ensure a complete implementation:
25 |
26 | ### 1. Tool Module Implementation
27 |
28 | ```python
29 | # Create a new file or modify an existing module in src/servicenow_mcp/tools/
30 |
31 | """
32 | {TOOL_NAME} tools for the ServiceNow MCP server.
33 |
34 | This module provides tools for {TOOL_DESCRIPTION}.
35 | """
36 |
37 | import logging
38 | from typing import Optional, List
39 |
40 | import requests
41 | from pydantic import BaseModel, Field
42 |
43 | from servicenow_mcp.auth.auth_manager import AuthManager
44 | from servicenow_mcp.utils.config import ServerConfig
45 |
46 | logger = logging.getLogger(__name__)
47 |
48 |
49 | class {ToolName}Params(BaseModel):
50 | """Parameters for {tool_name}."""
51 |
52 | # Define parameters with type annotations and descriptions
53 | param1: str = Field(..., description="Description of parameter 1")
54 | param2: Optional[str] = Field(None, description="Description of parameter 2")
55 | # Add more parameters as needed
56 |
57 |
58 | class {ToolName}Response(BaseModel):
59 | """Response from {tool_name} operations."""
60 |
61 | success: bool = Field(..., description="Whether the operation was successful")
62 | message: str = Field(..., description="Message describing the result")
63 | # Add more response fields as needed
64 |
65 |
66 | def {tool_name}(
67 | config: ServerConfig,
68 | auth_manager: AuthManager,
69 | params: {ToolName}Params,
70 | ) -> {ToolName}Response:
71 | """
72 | {Tool description with detailed explanation}.
73 |
74 | Args:
75 | config: Server configuration.
76 | auth_manager: Authentication manager.
77 | params: Parameters for {tool_name}.
78 |
79 | Returns:
80 | Response with {description of response}.
81 | """
82 | api_url = f"{config.api_url}/table/{table_name}"
83 |
84 | # Build request data
85 | data = {
86 | # Map parameters to API request fields
87 | "field1": params.param1,
88 | }
89 |
90 | if params.param2:
91 | data["field2"] = params.param2
92 | # Add conditional fields as needed
93 |
94 | # Make request
95 | try:
96 | response = requests.post( # or get, patch, delete as appropriate
97 | api_url,
98 | json=data,
99 | headers=auth_manager.get_headers(),
100 | timeout=config.timeout,
101 | )
102 | response.raise_for_status()
103 |
104 | result = response.json().get("result", {})
105 |
106 | return {ToolName}Response(
107 | success=True,
108 | message="{Tool name} completed successfully",
109 | # Add more response fields from result as needed
110 | )
111 |
112 | except requests.RequestException as e:
113 | logger.error(f"Failed to {tool_name}: {e}")
114 | return {ToolName}Response(
115 | success=False,
116 | message=f"Failed to {tool_name}: {str(e)}",
117 | )
118 | ```
119 |
120 | ### 2. Update tools/__init__.py
121 |
122 | ```python
123 | # Add import for new tool
124 | from servicenow_mcp.tools.{tool_module} import (
125 | {tool_name},
126 | )
127 |
128 | # Add to __all__ list
129 | __all__ = [
130 | # Existing tools...
131 |
132 | # New tools
133 | "{tool_name}",
134 | ]
135 | ```
136 |
137 | ### 3. Update server.py
138 |
139 | ```python
140 | # Add imports for the tool parameters and function
141 | from servicenow_mcp.tools.{tool_module} import (
142 | {ToolName}Params,
143 | )
144 | from servicenow_mcp.tools.{tool_module} import (
145 | {tool_name} as {tool_name}_tool,
146 | )
147 |
148 | # In the _register_tools method, add:
149 | @self.mcp_server.tool()
150 | def {tool_name}(params: {ToolName}Params) -> Dict[str, Any]:
151 | return {tool_name}_tool(
152 | self.config,
153 | self.auth_manager,
154 | params,
155 | )
156 | ```
157 |
158 | ### 4. Add Unit Tests
159 |
160 | ```python
161 | # Add to existing test file or create a new one in tests/test_{tool_module}.py
162 |
163 | @patch("requests.post") # Or appropriate HTTP method
164 | def test_{tool_name}(self, mock_post):
165 | """Test {tool_name} function."""
166 | # Configure mock
167 | mock_response = MagicMock()
168 | mock_response.raise_for_status = MagicMock()
169 | mock_response.json.return_value = {
170 | "result": {
171 | # Mocked response data
172 | }
173 | }
174 | mock_post.return_value = mock_response
175 |
176 | # Create test params
177 | params = {ToolName}Params(
178 | param1="value1",
179 | param2="value2",
180 | )
181 |
182 | # Call function
183 | result = {tool_name}(self.config, self.auth_manager, params)
184 |
185 | # Verify result
186 | self.assertTrue(result.success)
187 | # Add more assertions
188 |
189 | # Verify mock was called correctly
190 | mock_post.assert_called_once()
191 | call_args = mock_post.call_args
192 | self.assertEqual(call_args[0][0], f"{self.config.api_url}/table/{table_name}")
193 | # Add more assertions for request data
194 | ```
195 |
196 | ### 5. Update Documentation
197 |
198 | Create or update a markdown file in `docs/` that explains the tool:
199 |
200 | ```markdown
201 | # {Tool Category} in ServiceNow MCP
202 |
203 | ## {Tool Name}
204 |
205 | {Detailed description of the tool}
206 |
207 | ### Parameters
208 |
209 | | Parameter | Type | Required | Description |
210 | |-----------|------|----------|-------------|
211 | | param1 | string | Yes | Description of parameter 1 |
212 | | param2 | string | No | Description of parameter 2 |
213 | | ... | ... | ... | ... |
214 |
215 | ### Example
216 |
217 | ```python
218 | # Example usage of {tool_name}
219 | result = {tool_name}({
220 | "param1": "value1",
221 | "param2": "value2"
222 | })
223 | ```
224 |
225 | ### Response
226 |
227 | The tool returns a response with the following fields:
228 |
229 | | Field | Type | Description |
230 | |-------|------|-------------|
231 | | success | boolean | Whether the operation was successful |
232 | | message | string | A message describing the result |
233 | | ... | ... | ... |
234 | ```
235 |
236 | ### 6. Update README.md
237 |
238 | Add the new tool to the appropriate section in README.md:
239 |
240 | ```markdown
241 | #### {Tool Category} Tools
242 |
243 | 1. **existing_tool** - Description of existing tool
244 | ...
245 | N. **{tool_name}** - {Brief description of the new tool}
246 | ```
247 |
248 | Also add example usage to the "Example Usage with Claude" section:
249 |
250 | ```markdown
251 | #### {Tool Category} Examples
252 | - "Existing example query"
253 | ...
254 | - "{Example natural language query that would use the new tool}"
255 | ```
256 |
257 | ## Testing Guidelines
258 |
259 | 1. Write unit tests for all new functions and edge cases
260 | 2. Mock all external API calls
261 | 3. Test failure conditions and error handling
262 | 4. Verify parameter validation
263 |
264 | ## Documentation Guidelines
265 |
266 | 1. Use clear, concise language
267 | 2. Include all parameters and their descriptions
268 | 3. Provide usage examples
269 | 4. Document common errors and troubleshooting steps
270 | 5. Update README.md to showcase the new functionality
271 |
272 | ## Best Practices
273 |
274 | 1. Follow existing code patterns and style
275 | 2. Add appropriate error handling
276 | 3. Include detailed logging
277 | 4. Use meaningful variable and function names
278 | 5. Add type hints and docstrings
279 | 6. Keep functions focused and single-purpose
280 |
281 | ## Example Natural Language Commands for Claude
282 |
283 | List examples of natural language prompts that users can give to Claude that would trigger the new tool:
284 |
285 | - "Prompt example 1"
286 | - "Prompt example 2"
287 | - "Prompt example 3"
```
--------------------------------------------------------------------------------
/docs/catalog.md:
--------------------------------------------------------------------------------
```markdown
1 | # ServiceNow Service Catalog Integration
2 |
3 | This document provides information about the ServiceNow Service Catalog integration in the ServiceNow MCP server.
4 |
5 | ## Overview
6 |
7 | The ServiceNow Service Catalog integration allows you to:
8 |
9 | - List service catalog categories
10 | - List service catalog items
11 | - Get detailed information about specific catalog items, including their variables
12 | - Filter catalog items by category or search query
13 |
14 | ## Tools
15 |
16 | The following tools are available for interacting with the ServiceNow Service Catalog:
17 |
18 | ### `list_catalog_categories`
19 |
20 | Lists available service catalog categories.
21 |
22 | **Parameters:**
23 | - `limit` (int, default: 10): Maximum number of categories to return
24 | - `offset` (int, default: 0): Offset for pagination
25 | - `query` (string, optional): Search query for categories
26 | - `active` (boolean, default: true): Whether to only return active categories
27 |
28 | **Example:**
29 | ```python
30 | from servicenow_mcp.tools.catalog_tools import ListCatalogCategoriesParams, list_catalog_categories
31 |
32 | params = ListCatalogCategoriesParams(
33 | limit=5,
34 | query="hardware"
35 | )
36 | result = list_catalog_categories(config, auth_manager, params)
37 | ```
38 |
39 | ### `create_catalog_category`
40 |
41 | Creates a new service catalog category.
42 |
43 | **Parameters:**
44 | - `title` (string, required): Title of the category
45 | - `description` (string, optional): Description of the category
46 | - `parent` (string, optional): Parent category sys_id
47 | - `icon` (string, optional): Icon for the category
48 | - `active` (boolean, default: true): Whether the category is active
49 | - `order` (integer, optional): Order of the category
50 |
51 | **Example:**
52 | ```python
53 | from servicenow_mcp.tools.catalog_tools import CreateCatalogCategoryParams, create_catalog_category
54 |
55 | params = CreateCatalogCategoryParams(
56 | title="Cloud Services",
57 | description="Cloud-based services and resources",
58 | parent="parent_category_id",
59 | icon="cloud"
60 | )
61 | result = create_catalog_category(config, auth_manager, params)
62 | ```
63 |
64 | ### `update_catalog_category`
65 |
66 | Updates an existing service catalog category.
67 |
68 | **Parameters:**
69 | - `category_id` (string, required): Category ID or sys_id
70 | - `title` (string, optional): Title of the category
71 | - `description` (string, optional): Description of the category
72 | - `parent` (string, optional): Parent category sys_id
73 | - `icon` (string, optional): Icon for the category
74 | - `active` (boolean, optional): Whether the category is active
75 | - `order` (integer, optional): Order of the category
76 |
77 | **Example:**
78 | ```python
79 | from servicenow_mcp.tools.catalog_tools import UpdateCatalogCategoryParams, update_catalog_category
80 |
81 | params = UpdateCatalogCategoryParams(
82 | category_id="category123",
83 | title="IT Equipment",
84 | description="Updated description for IT equipment"
85 | )
86 | result = update_catalog_category(config, auth_manager, params)
87 | ```
88 |
89 | ### `move_catalog_items`
90 |
91 | Moves catalog items to a different category.
92 |
93 | **Parameters:**
94 | - `item_ids` (list of strings, required): List of catalog item IDs to move
95 | - `target_category_id` (string, required): Target category ID to move items to
96 |
97 | **Example:**
98 | ```python
99 | from servicenow_mcp.tools.catalog_tools import MoveCatalogItemsParams, move_catalog_items
100 |
101 | params = MoveCatalogItemsParams(
102 | item_ids=["item1", "item2", "item3"],
103 | target_category_id="target_category_id"
104 | )
105 | result = move_catalog_items(config, auth_manager, params)
106 | ```
107 |
108 | ### `list_catalog_items`
109 |
110 | Lists available service catalog items.
111 |
112 | **Parameters:**
113 | - `limit` (int, default: 10): Maximum number of items to return
114 | - `offset` (int, default: 0): Offset for pagination
115 | - `category` (string, optional): Filter by category
116 | - `query` (string, optional): Search query for items
117 | - `active` (boolean, default: true): Whether to only return active items
118 |
119 | **Example:**
120 | ```python
121 | from servicenow_mcp.tools.catalog_tools import ListCatalogItemsParams, list_catalog_items
122 |
123 | params = ListCatalogItemsParams(
124 | limit=5,
125 | category="hardware",
126 | query="laptop"
127 | )
128 | result = list_catalog_items(config, auth_manager, params)
129 | ```
130 |
131 | ### `get_catalog_item`
132 |
133 | Gets detailed information about a specific catalog item.
134 |
135 | **Parameters:**
136 | - `item_id` (string, required): Catalog item ID or sys_id
137 |
138 | **Example:**
139 | ```python
140 | from servicenow_mcp.tools.catalog_tools import GetCatalogItemParams, get_catalog_item
141 |
142 | params = GetCatalogItemParams(
143 | item_id="item123"
144 | )
145 | result = get_catalog_item(config, auth_manager, params)
146 | ```
147 |
148 | ## Resources
149 |
150 | The following resources are available for accessing the ServiceNow Service Catalog:
151 |
152 | ### `catalog://items`
153 |
154 | Lists service catalog items.
155 |
156 | **Example:**
157 | ```
158 | catalog://items
159 | ```
160 |
161 | ### `catalog://categories`
162 |
163 | Lists service catalog categories.
164 |
165 | **Example:**
166 | ```
167 | catalog://categories
168 | ```
169 |
170 | ### `catalog://{item_id}`
171 |
172 | Gets a specific catalog item by ID.
173 |
174 | **Example:**
175 | ```
176 | catalog://item123
177 | ```
178 |
179 | ## Integration with Claude Desktop
180 |
181 | To use the ServiceNow Service Catalog with Claude Desktop:
182 |
183 | 1. Configure the ServiceNow MCP server in Claude Desktop
184 | 2. Ask Claude questions about the service catalog
185 |
186 | **Example prompts:**
187 | - "Can you list the available service catalog categories in ServiceNow?"
188 | - "Can you show me the available items in the ServiceNow service catalog?"
189 | - "Can you list the catalog items in the Hardware category?"
190 | - "Can you show me the details of the 'New Laptop' catalog item?"
191 | - "Can you find catalog items related to 'software' in ServiceNow?"
192 | - "Can you create a new category called 'Cloud Services' in the service catalog?"
193 | - "Can you update the 'Hardware' category to rename it to 'IT Equipment'?"
194 | - "Can you move the 'Virtual Machine' catalog item to the 'Cloud Services' category?"
195 | - "Can you create a subcategory called 'Monitors' under the 'IT Equipment' category?"
196 | - "Can you reorganize our catalog by moving all software items to the 'Software' category?"
197 |
198 | ## Example Scripts
199 |
200 | ### Integration Test
201 |
202 | The `examples/catalog_integration_test.py` script demonstrates how to use the catalog tools directly:
203 |
204 | ```bash
205 | python examples/catalog_integration_test.py
206 | ```
207 |
208 | ### Claude Desktop Demo
209 |
210 | The `examples/claude_catalog_demo.py` script demonstrates how to use the catalog functionality with Claude Desktop:
211 |
212 | ```bash
213 | python examples/claude_catalog_demo.py
214 | ```
215 |
216 | ## Data Models
217 |
218 | ### CatalogItemModel
219 |
220 | Represents a ServiceNow catalog item.
221 |
222 | **Fields:**
223 | - `sys_id` (string): Unique identifier for the catalog item
224 | - `name` (string): Name of the catalog item
225 | - `short_description` (string, optional): Short description of the catalog item
226 | - `description` (string, optional): Detailed description of the catalog item
227 | - `category` (string, optional): Category of the catalog item
228 | - `price` (string, optional): Price of the catalog item
229 | - `picture` (string, optional): Picture URL of the catalog item
230 | - `active` (boolean, optional): Whether the catalog item is active
231 | - `order` (integer, optional): Order of the catalog item in its category
232 |
233 | ### CatalogCategoryModel
234 |
235 | Represents a ServiceNow catalog category.
236 |
237 | **Fields:**
238 | - `sys_id` (string): Unique identifier for the category
239 | - `title` (string): Title of the category
240 | - `description` (string, optional): Description of the category
241 | - `parent` (string, optional): Parent category ID
242 | - `icon` (string, optional): Icon of the category
243 | - `active` (boolean, optional): Whether the category is active
244 | - `order` (integer, optional): Order of the category
245 |
246 | ### CatalogItemVariableModel
247 |
248 | Represents a ServiceNow catalog item variable.
249 |
250 | **Fields:**
251 | - `sys_id` (string): Unique identifier for the variable
252 | - `name` (string): Name of the variable
253 | - `label` (string): Label of the variable
254 | - `type` (string): Type of the variable
255 | - `mandatory` (boolean, optional): Whether the variable is mandatory
256 | - `default_value` (string, optional): Default value of the variable
257 | - `help_text` (string, optional): Help text for the variable
258 | - `order` (integer, optional): Order of the variable
```
--------------------------------------------------------------------------------
/tests/test_changeset_resources.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the changeset resources.
3 |
4 | This module contains tests for the changeset resources in the ServiceNow MCP server.
5 | """
6 |
7 | import json
8 | import unittest
9 | import requests
10 | from unittest.mock import MagicMock, patch
11 |
12 | from servicenow_mcp.auth.auth_manager import AuthManager
13 | from servicenow_mcp.resources.changesets import ChangesetListParams, ChangesetResource
14 | from servicenow_mcp.utils.config import ServerConfig, AuthConfig, AuthType, BasicAuthConfig
15 |
16 |
17 | class TestChangesetResource(unittest.IsolatedAsyncioTestCase):
18 | """Tests for the changeset resource."""
19 |
20 | def setUp(self):
21 | """Set up test fixtures."""
22 | auth_config = AuthConfig(
23 | type=AuthType.BASIC,
24 | basic=BasicAuthConfig(
25 | username="test_user",
26 | password="test_password"
27 | )
28 | )
29 | self.server_config = ServerConfig(
30 | instance_url="https://test.service-now.com",
31 | auth=auth_config,
32 | )
33 | self.auth_manager = MagicMock(spec=AuthManager)
34 | self.auth_manager.get_headers.return_value = {"Authorization": "Bearer test"}
35 | self.changeset_resource = ChangesetResource(self.server_config, self.auth_manager)
36 |
37 | @patch("servicenow_mcp.resources.changesets.requests.get")
38 | async def test_list_changesets(self, mock_get):
39 | """Test listing changesets."""
40 | # Mock response
41 | mock_response = MagicMock()
42 | mock_response.text = json.dumps({
43 | "result": [
44 | {
45 | "sys_id": "123",
46 | "name": "Test Changeset",
47 | "state": "in_progress",
48 | "application": "Test App",
49 | "developer": "test.user",
50 | }
51 | ]
52 | })
53 | mock_response.raise_for_status.return_value = None
54 | mock_get.return_value = mock_response
55 |
56 | # Call the function
57 | params = ChangesetListParams(
58 | limit=10,
59 | offset=0,
60 | state="in_progress",
61 | application="Test App",
62 | developer="test.user",
63 | )
64 | result = await self.changeset_resource.list_changesets(params)
65 |
66 | # Verify the result
67 | result_json = json.loads(result)
68 | self.assertEqual(len(result_json["result"]), 1)
69 | self.assertEqual(result_json["result"][0]["sys_id"], "123")
70 | self.assertEqual(result_json["result"][0]["name"], "Test Changeset")
71 |
72 | # Verify the API call
73 | mock_get.assert_called_once()
74 | args, kwargs = mock_get.call_args
75 | self.assertEqual(args[0], "https://test.service-now.com/api/now/table/sys_update_set")
76 | self.assertEqual(kwargs["headers"], {"Authorization": "Bearer test"})
77 | self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
78 | self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
79 | self.assertIn("sysparm_query", kwargs["params"])
80 | self.assertIn("state=in_progress", kwargs["params"]["sysparm_query"])
81 | self.assertIn("application=Test App", kwargs["params"]["sysparm_query"])
82 | self.assertIn("developer=test.user", kwargs["params"]["sysparm_query"])
83 |
84 | @patch("servicenow_mcp.resources.changesets.requests.get")
85 | async def test_get_changeset(self, mock_get):
86 | """Test getting a changeset."""
87 | # Mock responses
88 | mock_changeset_response = MagicMock()
89 | mock_changeset_response.json.return_value = {
90 | "result": {
91 | "sys_id": "123",
92 | "name": "Test Changeset",
93 | "state": "in_progress",
94 | "application": "Test App",
95 | "developer": "test.user",
96 | }
97 | }
98 | mock_changeset_response.raise_for_status.return_value = None
99 |
100 | mock_changes_response = MagicMock()
101 | mock_changes_response.json.return_value = {
102 | "result": [
103 | {
104 | "sys_id": "456",
105 | "name": "test_file.py",
106 | "type": "file",
107 | "update_set": "123",
108 | }
109 | ]
110 | }
111 | mock_changes_response.raise_for_status.return_value = None
112 |
113 | # Set up the mock to return different responses for different URLs
114 | def side_effect(*args, **kwargs):
115 | url = args[0]
116 | if "sys_update_set" in url:
117 | return mock_changeset_response
118 | elif "sys_update_xml" in url:
119 | return mock_changes_response
120 | return None
121 |
122 | mock_get.side_effect = side_effect
123 |
124 | # Call the function
125 | result = await self.changeset_resource.get_changeset("123")
126 |
127 | # Verify the result
128 | result_json = json.loads(result)
129 | self.assertEqual(result_json["changeset"]["sys_id"], "123")
130 | self.assertEqual(result_json["changeset"]["name"], "Test Changeset")
131 | self.assertEqual(len(result_json["changes"]), 1)
132 | self.assertEqual(result_json["changes"][0]["sys_id"], "456")
133 | self.assertEqual(result_json["changes"][0]["name"], "test_file.py")
134 | self.assertEqual(result_json["change_count"], 1)
135 |
136 | # Verify the API calls
137 | self.assertEqual(mock_get.call_count, 2)
138 | first_call_args, first_call_kwargs = mock_get.call_args_list[0]
139 | self.assertEqual(
140 | first_call_args[0], "https://test.service-now.com/api/now/table/sys_update_set/123"
141 | )
142 | self.assertEqual(first_call_kwargs["headers"], {"Authorization": "Bearer test"})
143 |
144 | second_call_args, second_call_kwargs = mock_get.call_args_list[1]
145 | self.assertEqual(
146 | second_call_args[0], "https://test.service-now.com/api/now/table/sys_update_xml"
147 | )
148 | self.assertEqual(second_call_kwargs["headers"], {"Authorization": "Bearer test"})
149 | self.assertEqual(second_call_kwargs["params"]["sysparm_query"], "update_set=123")
150 |
151 | @patch("servicenow_mcp.resources.changesets.requests.get")
152 | async def test_list_changesets_error(self, mock_get):
153 | """Test listing changesets with an error."""
154 | # Mock response
155 | mock_get.side_effect = requests.exceptions.RequestException("Test error")
156 |
157 | # Call the function
158 | params = ChangesetListParams()
159 | result = await self.changeset_resource.list_changesets(params)
160 |
161 | # Verify the result
162 | result_json = json.loads(result)
163 | self.assertIn("error", result_json)
164 | self.assertIn("Test error", result_json["error"])
165 |
166 | @patch("servicenow_mcp.resources.changesets.requests.get")
167 | async def test_get_changeset_error(self, mock_get):
168 | """Test getting a changeset with an error."""
169 | # Mock response
170 | mock_get.side_effect = requests.exceptions.RequestException("Test error")
171 |
172 | # Call the function
173 | result = await self.changeset_resource.get_changeset("123")
174 |
175 | # Verify the result
176 | result_json = json.loads(result)
177 | self.assertIn("error", result_json)
178 | self.assertIn("Test error", result_json["error"])
179 |
180 |
181 | class TestChangesetListParams(unittest.TestCase):
182 | """Tests for the ChangesetListParams class."""
183 |
184 | def test_changeset_list_params(self):
185 | """Test ChangesetListParams."""
186 | params = ChangesetListParams(
187 | limit=20,
188 | offset=10,
189 | state="in_progress",
190 | application="Test App",
191 | developer="test.user",
192 | )
193 | self.assertEqual(params.limit, 20)
194 | self.assertEqual(params.offset, 10)
195 | self.assertEqual(params.state, "in_progress")
196 | self.assertEqual(params.application, "Test App")
197 | self.assertEqual(params.developer, "test.user")
198 |
199 | def test_changeset_list_params_defaults(self):
200 | """Test ChangesetListParams defaults."""
201 | params = ChangesetListParams()
202 | self.assertEqual(params.limit, 10)
203 | self.assertEqual(params.offset, 0)
204 | self.assertIsNone(params.state)
205 | self.assertIsNone(params.application)
206 | self.assertIsNone(params.developer)
207 |
208 |
209 | if __name__ == "__main__":
210 | unittest.main()
```
--------------------------------------------------------------------------------
/examples/catalog_optimization_example.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Example script demonstrating how to use the ServiceNow MCP catalog optimization tools.
4 |
5 | This script shows how to:
6 | 1. Get optimization recommendations for a ServiceNow Service Catalog
7 | 2. Update catalog items with improved descriptions
8 |
9 | Usage:
10 | python catalog_optimization_example.py [--update-descriptions]
11 |
12 | Options:
13 | --update-descriptions Automatically update items with poor descriptions
14 | """
15 |
16 | import argparse
17 | import logging
18 | import sys
19 | from typing import Dict
20 |
21 | from servicenow_mcp.server import ServiceNowMCP
22 | from servicenow_mcp.tools.catalog_optimization import (
23 | OptimizationRecommendationsParams,
24 | UpdateCatalogItemParams,
25 | )
26 | from servicenow_mcp.utils.config import load_config
27 |
28 | # Configure logging
29 | logging.basicConfig(
30 | level=logging.INFO,
31 | format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
32 | )
33 | logger = logging.getLogger(__name__)
34 |
35 |
36 | def get_optimization_recommendations(server: ServiceNowMCP) -> Dict:
37 | """
38 | Get optimization recommendations for the ServiceNow Service Catalog.
39 |
40 | Args:
41 | server: The ServiceNowMCP server instance
42 |
43 | Returns:
44 | Dict containing the optimization recommendations
45 | """
46 | logger.info("Getting catalog optimization recommendations...")
47 |
48 | # Create parameters for all recommendation types
49 | params = OptimizationRecommendationsParams(
50 | recommendation_types=[
51 | "inactive_items",
52 | "low_usage",
53 | "high_abandonment",
54 | "slow_fulfillment",
55 | "description_quality",
56 | ]
57 | )
58 |
59 | # Call the tool
60 | result = server.tools["get_optimization_recommendations"](params)
61 |
62 | if not result["success"]:
63 | logger.error(f"Failed to get optimization recommendations: {result.get('message', 'Unknown error')}")
64 | return {}
65 |
66 | return result
67 |
68 |
69 | def print_recommendations(recommendations: Dict) -> None:
70 | """
71 | Print the optimization recommendations in a readable format.
72 |
73 | Args:
74 | recommendations: The optimization recommendations dictionary
75 | """
76 | if not recommendations or "recommendations" not in recommendations:
77 | logger.warning("No recommendations available")
78 | return
79 |
80 | print("\n" + "=" * 80)
81 | print("SERVICENOW CATALOG OPTIMIZATION RECOMMENDATIONS")
82 | print("=" * 80)
83 |
84 | for rec in recommendations["recommendations"]:
85 | print(f"\n{rec['title']} ({rec['type']})")
86 | print("-" * len(rec['title']))
87 | print(f"Description: {rec['description']}")
88 | print(f"Impact: {rec['impact'].upper()}")
89 | print(f"Effort: {rec['effort'].upper()}")
90 | print(f"Recommended Action: {rec['action']}")
91 |
92 | if rec["items"]:
93 | print("\nAffected Items:")
94 | for i, item in enumerate(rec["items"], 1):
95 | print(f" {i}. {item['name']}")
96 | print(f" ID: {item['sys_id']}")
97 | print(f" Description: {item['short_description'] or '(No description)'}")
98 |
99 | # Print additional details based on recommendation type
100 | if rec["type"] == "low_usage":
101 | print(f" Order Count: {item['order_count']}")
102 | elif rec["type"] == "high_abandonment":
103 | print(f" Abandonment Rate: {item['abandonment_rate']}%")
104 | print(f" Cart Adds: {item['cart_adds']}")
105 | print(f" Completed Orders: {item['orders']}")
106 | elif rec["type"] == "slow_fulfillment":
107 | print(f" Avg. Fulfillment Time: {item['avg_fulfillment_time']} days")
108 | print(f" Compared to Catalog Avg: {item['avg_fulfillment_time_vs_catalog']}x slower")
109 | elif rec["type"] == "description_quality":
110 | print(f" Description Quality Score: {item['description_quality']}/100")
111 | print(f" Issues: {', '.join(item['quality_issues'])}")
112 |
113 | print()
114 | else:
115 | print("\nNo items found in this category.")
116 |
117 | print("=" * 80)
118 |
119 |
120 | def update_poor_descriptions(server: ServiceNowMCP, recommendations: Dict) -> None:
121 | """
122 | Update catalog items with poor descriptions.
123 |
124 | Args:
125 | server: The ServiceNowMCP server instance
126 | recommendations: The optimization recommendations dictionary
127 | """
128 | # Find the description quality recommendation
129 | description_rec = None
130 | for rec in recommendations.get("recommendations", []):
131 | if rec["type"] == "description_quality":
132 | description_rec = rec
133 | break
134 |
135 | if not description_rec or not description_rec.get("items"):
136 | logger.warning("No items with poor descriptions found")
137 | return
138 |
139 | logger.info(f"Found {len(description_rec['items'])} items with poor descriptions")
140 |
141 | # Update each item with a better description
142 | for item in description_rec["items"]:
143 | # Generate an improved description based on the item name and category
144 | improved_description = generate_improved_description(item)
145 |
146 | logger.info(f"Updating description for item: {item['name']} (ID: {item['sys_id']})")
147 | logger.info(f" Original: {item['short_description'] or '(No description)'}")
148 | logger.info(f" Improved: {improved_description}")
149 |
150 | # Create parameters for updating the item
151 | params = UpdateCatalogItemParams(
152 | item_id=item["sys_id"],
153 | short_description=improved_description,
154 | )
155 |
156 | # Call the tool
157 | result = server.tools["update_catalog_item"](params)
158 |
159 | if result["success"]:
160 | logger.info(f"Successfully updated description for {item['name']}")
161 | else:
162 | logger.error(f"Failed to update description: {result.get('message', 'Unknown error')}")
163 |
164 |
165 | def generate_improved_description(item: Dict) -> str:
166 | """
167 | Generate an improved description for a catalog item.
168 |
169 | In a real implementation, this could use AI to generate better descriptions,
170 | but for this example we'll use a simple template-based approach.
171 |
172 | Args:
173 | item: The catalog item dictionary
174 |
175 | Returns:
176 | An improved description string
177 | """
178 | name = item["name"]
179 | category = item.get("category", "").lower()
180 |
181 | # Simple templates based on category
182 | if "hardware" in category:
183 | return f"Enterprise-grade {name.lower()} for professional use. Includes standard warranty and IT support."
184 | elif "software" in category:
185 | return f"Licensed {name.lower()} application with full feature set. Includes installation support."
186 | elif "service" in category:
187 | return f"Professional {name.lower()} service delivered by our expert team. Includes consultation and implementation."
188 | else:
189 | return f"High-quality {name.lower()} available through IT self-service. Contact the service desk for assistance."
190 |
191 |
192 | def main():
193 | """Main function to run the example."""
194 | parser = argparse.ArgumentParser(description="ServiceNow Catalog Optimization Example")
195 | parser.add_argument(
196 | "--update-descriptions",
197 | action="store_true",
198 | help="Automatically update items with poor descriptions",
199 | )
200 | args = parser.parse_args()
201 |
202 | try:
203 | # Load configuration
204 | config = load_config()
205 |
206 | # Initialize the ServiceNow MCP server
207 | server = ServiceNowMCP(config)
208 |
209 | # Get optimization recommendations
210 | recommendations = get_optimization_recommendations(server)
211 |
212 | # Print the recommendations
213 | print_recommendations(recommendations)
214 |
215 | # Update poor descriptions if requested
216 | if args.update_descriptions and recommendations:
217 | update_poor_descriptions(server, recommendations)
218 |
219 | except Exception as e:
220 | logger.exception(f"Error running catalog optimization example: {e}")
221 | sys.exit(1)
222 |
223 |
224 | if __name__ == "__main__":
225 | main()
```
--------------------------------------------------------------------------------
/scripts/setup_oauth.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """
3 | ServiceNow OAuth Setup Script
4 |
5 | This script helps set up and test OAuth authentication with ServiceNow.
6 | It will:
7 | 1. Get an OAuth token using client credentials
8 | 2. Test the token with a simple API call
9 | 3. Update the .env file with the OAuth configuration
10 |
11 | Usage:
12 | python scripts/setup_oauth.py
13 | """
14 |
15 | import os
16 | import sys
17 | import json
18 | import requests
19 | import base64
20 | from pathlib import Path
21 | from dotenv import load_dotenv
22 |
23 | # Add the project root to the Python path
24 | sys.path.insert(0, str(Path(__file__).parent.parent))
25 |
26 | def setup_oauth():
27 | # Load environment variables
28 | load_dotenv()
29 |
30 | # Get ServiceNow instance URL
31 | instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
32 | if not instance_url or instance_url == "https://your-instance.service-now.com":
33 | instance_url = input("Enter your ServiceNow instance URL (e.g., https://dev296866.service-now.com): ")
34 |
35 | print("\n=== ServiceNow OAuth Setup ===")
36 | print("This script will help you set up OAuth authentication for your ServiceNow instance.")
37 | print("You'll need to create an OAuth API client in ServiceNow first.")
38 | print("\nTo create an OAuth API client:")
39 | print("1. Log in to your ServiceNow instance")
40 | print("2. Navigate to System OAuth > Application Registry")
41 | print("3. Click 'New'")
42 | print("4. Select 'Create an OAuth API endpoint for external clients'")
43 | print("5. Fill in the required fields:")
44 | print(" - Name: MCP Client (or any name you prefer)")
45 | print(" - Redirect URL: http://localhost (for testing)")
46 | print(" - Active: Checked")
47 | print(" - Refresh Token Lifespan: 8 hours (or your preference)")
48 | print(" - Access Token Lifespan: 30 minutes (or your preference)")
49 | print("6. Save the application and note down the Client ID and Client Secret")
50 | print("7. Go to the 'OAuth Scopes' related list and add appropriate scopes (e.g., 'admin')")
51 |
52 | # Get OAuth credentials
53 | client_id = input("\nEnter your Client ID: ")
54 | client_secret = input("Enter your Client Secret: ")
55 |
56 | # Get username and password for resource owner grant
57 | username = os.getenv("SERVICENOW_USERNAME")
58 | password = os.getenv("SERVICENOW_PASSWORD")
59 |
60 | if not username or username == "your-username":
61 | username = input("Enter your ServiceNow username: ")
62 |
63 | if not password or password == "your-password":
64 | password = input("Enter your ServiceNow password: ")
65 |
66 | # Set token URL
67 | token_url = f"{instance_url}/oauth_token.do"
68 |
69 | print(f"\nTesting OAuth connection to {instance_url}...")
70 | print("Trying different OAuth grant types...")
71 |
72 | # Try different OAuth grant types
73 | access_token = None
74 |
75 | # 1. Try client credentials grant
76 | try:
77 | print("\nAttempting client_credentials grant...")
78 | # Create authorization header
79 | auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
80 |
81 | token_response = requests.post(
82 | token_url,
83 | headers={
84 | "Authorization": f"Basic {auth_header}",
85 | "Content-Type": "application/x-www-form-urlencoded"
86 | },
87 | data={
88 | "grant_type": "client_credentials"
89 | }
90 | )
91 |
92 | if token_response.status_code == 200:
93 | token_data = token_response.json()
94 | access_token = token_data.get("access_token")
95 | print("✅ Successfully obtained OAuth token using client_credentials grant!")
96 | else:
97 | print(f"❌ Failed with client_credentials grant: {token_response.status_code}")
98 | print(f"Response: {token_response.text}")
99 | except Exception as e:
100 | print(f"❌ Error with client_credentials grant: {e}")
101 |
102 | # 2. Try password grant if client credentials failed
103 | if not access_token:
104 | try:
105 | print("\nAttempting password grant...")
106 | token_response = requests.post(
107 | token_url,
108 | headers={
109 | "Authorization": f"Basic {auth_header}",
110 | "Content-Type": "application/x-www-form-urlencoded"
111 | },
112 | data={
113 | "grant_type": "password",
114 | "username": username,
115 | "password": password
116 | }
117 | )
118 |
119 | if token_response.status_code == 200:
120 | token_data = token_response.json()
121 | access_token = token_data.get("access_token")
122 | print("✅ Successfully obtained OAuth token using password grant!")
123 | else:
124 | print(f"❌ Failed with password grant: {token_response.status_code}")
125 | print(f"Response: {token_response.text}")
126 | except Exception as e:
127 | print(f"❌ Error with password grant: {e}")
128 |
129 | # If we have a token, test it
130 | if access_token:
131 | # Test the token with a simple API call
132 | test_url = f"{instance_url}/api/now/table/incident?sysparm_limit=1"
133 | test_response = requests.get(
134 | test_url,
135 | headers={
136 | "Authorization": f"Bearer {access_token}",
137 | "Accept": "application/json"
138 | }
139 | )
140 |
141 | if test_response.status_code == 200:
142 | print("✅ Successfully tested OAuth token with API call!")
143 | data = test_response.json()
144 | print(f"Retrieved {len(data.get('result', []))} incident(s)")
145 |
146 | # Update .env file
147 | update_env = input("\nDo you want to update your .env file with these OAuth credentials? (y/n): ")
148 | if update_env.lower() == 'y':
149 | env_path = Path(__file__).parent.parent / '.env'
150 | with open(env_path, 'r') as f:
151 | env_lines = f.readlines()
152 |
153 | # Helper to update or insert a key
154 | def set_env_var(lines, key, value):
155 | found = False
156 | for i, line in enumerate(lines):
157 | if line.strip().startswith(f'{key}=') or line.strip().startswith(f'#{key}='):
158 | lines[i] = f'{key}={value}\n'
159 | found = True
160 | break
161 | if not found:
162 | lines.append(f'{key}={value}\n')
163 | return lines
164 |
165 | # Update or insert OAuth configuration
166 | env_lines = set_env_var(env_lines, 'SERVICENOW_AUTH_TYPE', 'oauth')
167 | env_lines = set_env_var(env_lines, 'SERVICENOW_CLIENT_ID', client_id)
168 | env_lines = set_env_var(env_lines, 'SERVICENOW_CLIENT_SECRET', client_secret)
169 | env_lines = set_env_var(env_lines, 'SERVICENOW_TOKEN_URL', token_url)
170 | env_lines = set_env_var(env_lines, 'SERVICENOW_USERNAME', username)
171 | env_lines = set_env_var(env_lines, 'SERVICENOW_PASSWORD', password)
172 |
173 | with open(env_path, 'w') as f:
174 | f.writelines(env_lines)
175 |
176 | print("✅ Updated .env file with OAuth configuration!")
177 | print("\nYou can now use OAuth authentication with the ServiceNow MCP server.")
178 | print("To test it, run: python scripts/test_connection.py")
179 |
180 | return True
181 | else:
182 | print(f"❌ Failed to test OAuth token with API call: {test_response.status_code}")
183 | print(f"Response: {test_response.text}")
184 | return False
185 | else:
186 | print("\n❌ Failed to obtain OAuth token with any grant type.")
187 | print("\nPossible issues:")
188 | print("1. The OAuth client may not have the correct scopes")
189 | print("2. The client ID or client secret may be incorrect")
190 | print("3. The OAuth client may not be active")
191 | print("4. The username/password may be incorrect")
192 | print("\nPlease check your ServiceNow instance OAuth configuration and try again.")
193 | return False
194 |
195 | if __name__ == "__main__":
196 | setup_oauth()
```
--------------------------------------------------------------------------------
/examples/workflow_management_demo.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Workflow Management Demo
4 |
5 | This script demonstrates how to use the ServiceNow MCP workflow management tools
6 | to view, create, and modify workflows in ServiceNow.
7 | """
8 |
9 | import os
10 | import sys
11 | import json
12 | from datetime import datetime
13 |
14 | from dotenv import load_dotenv
15 |
16 | # Add the parent directory to the path so we can import the package
17 | sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
18 |
19 | from servicenow_mcp.auth.auth_manager import AuthManager
20 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
21 | from servicenow_mcp.tools.workflow_tools import (
22 | list_workflows,
23 | get_workflow_details,
24 | list_workflow_versions,
25 | get_workflow_activities,
26 | create_workflow,
27 | update_workflow,
28 | activate_workflow,
29 | deactivate_workflow,
30 | add_workflow_activity,
31 | update_workflow_activity,
32 | delete_workflow_activity,
33 | reorder_workflow_activities,
34 | )
35 |
36 |
37 | def print_json(data):
38 | """Print JSON data in a readable format."""
39 | print(json.dumps(data, indent=2))
40 |
41 |
42 | def main():
43 | """Main function to demonstrate workflow management tools."""
44 | # Load environment variables from .env file
45 | load_dotenv()
46 |
47 | # Get ServiceNow credentials from environment variables
48 | instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
49 | username = os.getenv("SERVICENOW_USERNAME")
50 | password = os.getenv("SERVICENOW_PASSWORD")
51 |
52 | if not all([instance_url, username, password]):
53 | print("Error: Missing required environment variables.")
54 | print("Please set SERVICENOW_INSTANCE_URL, SERVICENOW_USERNAME, and SERVICENOW_PASSWORD.")
55 | sys.exit(1)
56 |
57 | # Create authentication configuration
58 | auth_config = AuthConfig(
59 | type=AuthType.BASIC,
60 | basic=BasicAuthConfig(username=username, password=password),
61 | )
62 |
63 | # Create server configuration
64 | server_config = ServerConfig(
65 | instance_url=instance_url,
66 | auth=auth_config,
67 | )
68 |
69 | # Create authentication manager
70 | auth_manager = AuthManager(auth_config)
71 |
72 | print("ServiceNow Workflow Management Demo")
73 | print("===================================")
74 | print(f"Instance URL: {instance_url}")
75 | print(f"Username: {username}")
76 | print()
77 |
78 | # List active workflows
79 | print("Listing active workflows...")
80 | workflows_result = list_workflows(
81 | auth_manager,
82 | server_config,
83 | {
84 | "limit": 5,
85 | "active": True,
86 | },
87 | )
88 | print_json(workflows_result)
89 | print()
90 |
91 | # Check if we have any workflows
92 | if workflows_result.get("count", 0) == 0:
93 | print("No active workflows found. Creating a new workflow...")
94 |
95 | # Create a new workflow
96 | new_workflow_result = create_workflow(
97 | auth_manager,
98 | server_config,
99 | {
100 | "name": f"Demo Workflow {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
101 | "description": "A demo workflow created by the ServiceNow MCP workflow management demo",
102 | "table": "incident",
103 | "active": True,
104 | },
105 | )
106 | print_json(new_workflow_result)
107 | print()
108 |
109 | if "error" in new_workflow_result:
110 | print("Error creating workflow. Exiting.")
111 | sys.exit(1)
112 |
113 | workflow_id = new_workflow_result["workflow"]["sys_id"]
114 | else:
115 | # Use the first workflow from the list
116 | workflow_id = workflows_result["workflows"][0]["sys_id"]
117 |
118 | print(f"Using workflow with ID: {workflow_id}")
119 | print()
120 |
121 | # Get workflow details
122 | print("Getting workflow details...")
123 | workflow_details = get_workflow_details(
124 | auth_manager,
125 | server_config,
126 | {
127 | "workflow_id": workflow_id,
128 | },
129 | )
130 | print_json(workflow_details)
131 | print()
132 |
133 | # List workflow versions
134 | print("Listing workflow versions...")
135 | versions_result = list_workflow_versions(
136 | auth_manager,
137 | server_config,
138 | {
139 | "workflow_id": workflow_id,
140 | "limit": 5,
141 | },
142 | )
143 | print_json(versions_result)
144 | print()
145 |
146 | # Get workflow activities
147 | print("Getting workflow activities...")
148 | activities_result = get_workflow_activities(
149 | auth_manager,
150 | server_config,
151 | {
152 | "workflow_id": workflow_id,
153 | },
154 | )
155 | print_json(activities_result)
156 | print()
157 |
158 | # Add a new activity to the workflow
159 | print("Adding a new approval activity to the workflow...")
160 | add_activity_result = add_workflow_activity(
161 | auth_manager,
162 | server_config,
163 | {
164 | "workflow_id": workflow_id,
165 | "name": f"Demo Approval {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
166 | "description": "A demo approval activity",
167 | "activity_type": "approval",
168 | },
169 | )
170 | print_json(add_activity_result)
171 | print()
172 |
173 | if "error" in add_activity_result:
174 | print("Error adding activity. Skipping activity modification steps.")
175 | else:
176 | activity_id = add_activity_result["activity"]["sys_id"]
177 |
178 | # Update the activity
179 | print("Updating the activity...")
180 | update_activity_result = update_workflow_activity(
181 | auth_manager,
182 | server_config,
183 | {
184 | "activity_id": activity_id,
185 | "name": f"Updated Demo Approval {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
186 | "description": "An updated demo approval activity",
187 | },
188 | )
189 | print_json(update_activity_result)
190 | print()
191 |
192 | # Get the updated activities
193 | print("Getting updated workflow activities...")
194 | updated_activities_result = get_workflow_activities(
195 | auth_manager,
196 | server_config,
197 | {
198 | "workflow_id": workflow_id,
199 | },
200 | )
201 | print_json(updated_activities_result)
202 | print()
203 |
204 | # If there are multiple activities, reorder them
205 | if updated_activities_result.get("count", 0) > 1:
206 | print("Reordering workflow activities...")
207 | activity_ids = [activity["sys_id"] for activity in updated_activities_result["activities"]]
208 | # Reverse the order
209 | activity_ids.reverse()
210 |
211 | reorder_result = reorder_workflow_activities(
212 | auth_manager,
213 | server_config,
214 | {
215 | "workflow_id": workflow_id,
216 | "activity_ids": activity_ids,
217 | },
218 | )
219 | print_json(reorder_result)
220 | print()
221 |
222 | # Get the reordered activities
223 | print("Getting reordered workflow activities...")
224 | reordered_activities_result = get_workflow_activities(
225 | auth_manager,
226 | server_config,
227 | {
228 | "workflow_id": workflow_id,
229 | },
230 | )
231 | print_json(reordered_activities_result)
232 | print()
233 |
234 | # Update the workflow
235 | print("Updating the workflow...")
236 | update_result = update_workflow(
237 | auth_manager,
238 | server_config,
239 | {
240 | "workflow_id": workflow_id,
241 | "description": f"Updated description {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
242 | },
243 | )
244 | print_json(update_result)
245 | print()
246 |
247 | # Deactivate the workflow
248 | print("Deactivating the workflow...")
249 | deactivate_result = deactivate_workflow(
250 | auth_manager,
251 | server_config,
252 | {
253 | "workflow_id": workflow_id,
254 | },
255 | )
256 | print_json(deactivate_result)
257 | print()
258 |
259 | # Activate the workflow
260 | print("Activating the workflow...")
261 | activate_result = activate_workflow(
262 | auth_manager,
263 | server_config,
264 | {
265 | "workflow_id": workflow_id,
266 | },
267 | )
268 | print_json(activate_result)
269 | print()
270 |
271 | print("Workflow management demo completed successfully!")
272 |
273 |
274 | if __name__ == "__main__":
275 | main()
```
--------------------------------------------------------------------------------
/scripts/test_connection.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """
3 | ServiceNow Connection Test Script
4 |
5 | This script tests the connection to a ServiceNow instance using the credentials
6 | provided in the .env file. It supports all authentication methods:
7 | - Basic authentication (username/password)
8 | - OAuth authentication (client ID/client secret)
9 | - API key authentication
10 |
11 | Usage:
12 | python scripts/test_connection.py
13 | """
14 |
15 | import os
16 | import sys
17 | import requests
18 | import base64
19 | from pathlib import Path
20 | from dotenv import load_dotenv
21 |
22 | # Add the project root to the Python path
23 | sys.path.insert(0, str(Path(__file__).parent.parent))
24 |
25 | def get_oauth_token(instance_url, client_id, client_secret, username=None, password=None):
26 | """Get an OAuth token from ServiceNow."""
27 | token_url = os.getenv("SERVICENOW_TOKEN_URL", f"{instance_url}/oauth_token.do")
28 |
29 | # Create authorization header
30 | auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
31 |
32 | # Try different OAuth grant types
33 | access_token = None
34 |
35 | # 1. Try client credentials grant
36 | try:
37 | print("Attempting client_credentials grant...")
38 | token_response = requests.post(
39 | token_url,
40 | headers={
41 | "Authorization": f"Basic {auth_header}",
42 | "Content-Type": "application/x-www-form-urlencoded"
43 | },
44 | data={
45 | "grant_type": "client_credentials"
46 | }
47 | )
48 |
49 | if token_response.status_code == 200:
50 | token_data = token_response.json()
51 | access_token = token_data.get("access_token")
52 | print("✅ Successfully obtained OAuth token using client_credentials grant!")
53 | return access_token
54 | else:
55 | print(f"❌ Failed with client_credentials grant: {token_response.status_code}")
56 | print(f"Response: {token_response.text}")
57 | except Exception as e:
58 | print(f"❌ Error with client_credentials grant: {e}")
59 |
60 | # 2. Try password grant if client credentials failed and we have username/password
61 | if not access_token and username and password:
62 | try:
63 | print("Attempting password grant...")
64 | token_response = requests.post(
65 | token_url,
66 | headers={
67 | "Authorization": f"Basic {auth_header}",
68 | "Content-Type": "application/x-www-form-urlencoded"
69 | },
70 | data={
71 | "grant_type": "password",
72 | "username": username,
73 | "password": password
74 | }
75 | )
76 |
77 | if token_response.status_code == 200:
78 | token_data = token_response.json()
79 | access_token = token_data.get("access_token")
80 | print("✅ Successfully obtained OAuth token using password grant!")
81 | return access_token
82 | else:
83 | print(f"❌ Failed with password grant: {token_response.status_code}")
84 | print(f"Response: {token_response.text}")
85 | except Exception as e:
86 | print(f"❌ Error with password grant: {e}")
87 |
88 | return None
89 |
90 | def test_connection():
91 | # Load environment variables
92 | load_dotenv()
93 |
94 | # Print all environment variables related to ServiceNow
95 | print("Environment variables:")
96 | for key, value in os.environ.items():
97 | if "SERVICENOW" in key:
98 | masked_value = value
99 | if "PASSWORD" in key or "SECRET" in key:
100 | masked_value = "*" * len(value)
101 | print(f" {key}={masked_value}")
102 | print()
103 |
104 | # Get ServiceNow credentials
105 | instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
106 | auth_type = os.getenv("SERVICENOW_AUTH_TYPE", "basic")
107 |
108 | # Check if instance URL is set
109 | if not instance_url or instance_url == "https://your-instance.service-now.com":
110 | print("Error: ServiceNow instance URL is not set or is using the default value.")
111 | print("Please update the SERVICENOW_INSTANCE_URL in your .env file.")
112 | sys.exit(1)
113 |
114 | print(f"Testing connection to ServiceNow instance: {instance_url}")
115 | print(f"Authentication type: {auth_type}")
116 |
117 | # Construct API endpoint URL
118 | api_url = f"{instance_url}/api/now/table/incident?sysparm_limit=1"
119 | headers = {"Accept": "application/json"}
120 | auth = None
121 |
122 | # Set up authentication based on auth type
123 | if auth_type == "basic":
124 | username = os.getenv("SERVICENOW_USERNAME")
125 | password = os.getenv("SERVICENOW_PASSWORD")
126 |
127 | if not username or not password or username == "your-username" or password == "your-password":
128 | print("Error: Username or password is not set or is using the default value.")
129 | print("Please update the SERVICENOW_USERNAME and SERVICENOW_PASSWORD in your .env file.")
130 | sys.exit(1)
131 |
132 | auth = (username, password)
133 | print("Using basic authentication (username/password)")
134 | print(f"Username: {username}")
135 | print(f"Password: {'*' * len(password)}")
136 |
137 | elif auth_type == "oauth":
138 | client_id = os.getenv("SERVICENOW_CLIENT_ID")
139 | client_secret = os.getenv("SERVICENOW_CLIENT_SECRET")
140 | username = os.getenv("SERVICENOW_USERNAME")
141 | password = os.getenv("SERVICENOW_PASSWORD")
142 |
143 | if not client_id or not client_secret or client_id == "your-client-id" or client_secret == "your-client-secret":
144 | print("Error: Client ID or Client Secret is not set or is using the default value.")
145 | print("Please update the SERVICENOW_CLIENT_ID and SERVICENOW_CLIENT_SECRET in your .env file.")
146 | print("You can run scripts/setup_oauth.py to set up OAuth authentication.")
147 | sys.exit(1)
148 |
149 | print("Using OAuth authentication (client ID/client secret)")
150 | access_token = get_oauth_token(instance_url, client_id, client_secret, username, password)
151 |
152 | if not access_token:
153 | print("Failed to obtain OAuth token. Please check your credentials.")
154 | sys.exit(1)
155 |
156 | headers["Authorization"] = f"Bearer {access_token}"
157 |
158 | elif auth_type == "api_key":
159 | api_key = os.getenv("SERVICENOW_API_KEY")
160 | api_key_header = os.getenv("SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key")
161 |
162 | if not api_key or api_key == "your-api-key":
163 | print("Error: API key is not set or is using the default value.")
164 | print("Please update the SERVICENOW_API_KEY in your .env file.")
165 | print("You can run scripts/setup_api_key.py to set up API key authentication.")
166 | sys.exit(1)
167 |
168 | print(f"Using API key authentication (header: {api_key_header})")
169 | headers[api_key_header] = api_key
170 |
171 | else:
172 | print(f"Error: Unsupported authentication type: {auth_type}")
173 | print("Supported types: basic, oauth, api_key")
174 | sys.exit(1)
175 |
176 | try:
177 | # Print request details
178 | print("\nRequest details:")
179 | print(f"URL: {api_url}")
180 | print(f"Headers: {headers}")
181 | if auth:
182 | print(f"Auth: ({auth[0]}, {'*' * len(auth[1])})")
183 |
184 | # Make a test request
185 | if auth:
186 | response = requests.get(api_url, auth=auth, headers=headers)
187 | else:
188 | response = requests.get(api_url, headers=headers)
189 |
190 | # Print response details
191 | print("\nResponse details:")
192 | print(f"Status code: {response.status_code}")
193 | print(f"Response headers: {dict(response.headers)}")
194 |
195 | # Check response
196 | if response.status_code == 200:
197 | print("\n✅ Connection successful!")
198 | data = response.json()
199 | print(f"Retrieved {len(data.get('result', []))} incident(s)")
200 | print("\nSample response:")
201 | print(f"{response.text[:500]}...")
202 | return True
203 | else:
204 | print(f"\n❌ Connection failed with status code: {response.status_code}")
205 | print(f"Response: {response.text}")
206 | return False
207 |
208 | except requests.exceptions.RequestException as e:
209 | print(f"\n❌ Connection error: {e}")
210 | return False
211 |
212 | if __name__ == "__main__":
213 | test_connection()
```
--------------------------------------------------------------------------------
/docs/workflow_management.md:
--------------------------------------------------------------------------------
```markdown
1 | # Workflow Management in ServiceNow MCP
2 |
3 | This document provides detailed information about the workflow management tools available in the ServiceNow MCP server.
4 |
5 | ## Overview
6 |
7 | 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.
8 |
9 | ## Available Tools
10 |
11 | ### Viewing Workflows
12 |
13 | 1. **list_workflows** - List workflows from ServiceNow
14 | - Parameters:
15 | - `limit` (optional): Maximum number of records to return (default: 10)
16 | - `offset` (optional): Offset to start from (default: 0)
17 | - `active` (optional): Filter by active status (true/false)
18 | - `name` (optional): Filter by name (contains)
19 | - `query` (optional): Additional query string
20 |
21 | 2. **get_workflow_details** - Get detailed information about a specific workflow
22 | - Parameters:
23 | - `workflow_id` (required): Workflow ID or sys_id
24 |
25 | 3. **list_workflow_versions** - List all versions of a specific workflow
26 | - Parameters:
27 | - `workflow_id` (required): Workflow ID or sys_id
28 | - `limit` (optional): Maximum number of records to return (default: 10)
29 | - `offset` (optional): Offset to start from (default: 0)
30 |
31 | 4. **get_workflow_activities** - Get all activities in a workflow
32 | - Parameters:
33 | - `workflow_id` (required): Workflow ID or sys_id
34 | - `version` (optional): Specific version to get activities for (if not provided, the latest published version will be used)
35 |
36 | ### Modifying Workflows
37 |
38 | 5. **create_workflow** - Create a new workflow in ServiceNow
39 | - Parameters:
40 | - `name` (required): Name of the workflow
41 | - `description` (optional): Description of the workflow
42 | - `table` (optional): Table the workflow applies to
43 | - `active` (optional): Whether the workflow is active (default: true)
44 | - `attributes` (optional): Additional attributes for the workflow
45 |
46 | 6. **update_workflow** - Update an existing workflow
47 | - Parameters:
48 | - `workflow_id` (required): Workflow ID or sys_id
49 | - `name` (optional): Name of the workflow
50 | - `description` (optional): Description of the workflow
51 | - `table` (optional): Table the workflow applies to
52 | - `active` (optional): Whether the workflow is active
53 | - `attributes` (optional): Additional attributes for the workflow
54 |
55 | 7. **activate_workflow** - Activate a workflow
56 | - Parameters:
57 | - `workflow_id` (required): Workflow ID or sys_id
58 |
59 | 8. **deactivate_workflow** - Deactivate a workflow
60 | - Parameters:
61 | - `workflow_id` (required): Workflow ID or sys_id
62 |
63 | ### Managing Workflow Activities
64 |
65 | 9. **add_workflow_activity** - Add a new activity to a workflow
66 | - Parameters:
67 | - `workflow_id` (required): Workflow ID or sys_id
68 | - `name` (required): Name of the activity
69 | - `description` (optional): Description of the activity
70 | - `activity_type` (required): Type of activity (e.g., 'approval', 'task', 'notification')
71 | - `attributes` (optional): Additional attributes for the activity
72 | - `position` (optional): Position in the workflow (if not provided, the activity will be added at the end)
73 |
74 | 10. **update_workflow_activity** - Update an existing activity in a workflow
75 | - Parameters:
76 | - `activity_id` (required): Activity ID or sys_id
77 | - `name` (optional): Name of the activity
78 | - `description` (optional): Description of the activity
79 | - `attributes` (optional): Additional attributes for the activity
80 |
81 | 11. **delete_workflow_activity** - Delete an activity from a workflow
82 | - Parameters:
83 | - `activity_id` (required): Activity ID or sys_id
84 |
85 | 12. **reorder_workflow_activities** - Change the order of activities in a workflow
86 | - Parameters:
87 | - `workflow_id` (required): Workflow ID or sys_id
88 | - `activity_ids` (required): List of activity IDs in the desired order
89 |
90 | ## Usage Examples
91 |
92 | ### Viewing Workflows
93 |
94 | #### List all active workflows
95 |
96 | ```python
97 | result = list_workflows({
98 | "active": True,
99 | "limit": 20
100 | })
101 | ```
102 |
103 | #### Get details about a specific workflow
104 |
105 | ```python
106 | result = get_workflow_details({
107 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
108 | })
109 | ```
110 |
111 | #### List all versions of a workflow
112 |
113 | ```python
114 | result = list_workflow_versions({
115 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
116 | })
117 | ```
118 |
119 | #### Get all activities in a workflow
120 |
121 | ```python
122 | result = get_workflow_activities({
123 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
124 | })
125 | ```
126 |
127 | ### Modifying Workflows
128 |
129 | #### Create a new workflow
130 |
131 | ```python
132 | result = create_workflow({
133 | "name": "Software License Request",
134 | "description": "Workflow for handling software license requests",
135 | "table": "sc_request"
136 | })
137 | ```
138 |
139 | #### Update an existing workflow
140 |
141 | ```python
142 | result = update_workflow({
143 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
144 | "description": "Updated workflow description",
145 | "active": True
146 | })
147 | ```
148 |
149 | #### Activate a workflow
150 |
151 | ```python
152 | result = activate_workflow({
153 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
154 | })
155 | ```
156 |
157 | #### Deactivate a workflow
158 |
159 | ```python
160 | result = deactivate_workflow({
161 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590"
162 | })
163 | ```
164 |
165 | ### Managing Workflow Activities
166 |
167 | #### Add a new activity to a workflow
168 |
169 | ```python
170 | result = add_workflow_activity({
171 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
172 | "name": "Manager Approval",
173 | "description": "Approval step for the manager",
174 | "activity_type": "approval"
175 | })
176 | ```
177 |
178 | #### Update an existing activity
179 |
180 | ```python
181 | result = update_workflow_activity({
182 | "activity_id": "3cda7cda87a9c150e0b0df23cebb3591",
183 | "name": "Updated Activity Name",
184 | "description": "Updated activity description"
185 | })
186 | ```
187 |
188 | #### Delete an activity
189 |
190 | ```python
191 | result = delete_workflow_activity({
192 | "activity_id": "3cda7cda87a9c150e0b0df23cebb3591"
193 | })
194 | ```
195 |
196 | #### Reorder activities in a workflow
197 |
198 | ```python
199 | result = reorder_workflow_activities({
200 | "workflow_id": "2bda7cda87a9c150e0b0df23cebb3590",
201 | "activity_ids": [
202 | "3cda7cda87a9c150e0b0df23cebb3591",
203 | "4cda7cda87a9c150e0b0df23cebb3592",
204 | "5cda7cda87a9c150e0b0df23cebb3593"
205 | ]
206 | })
207 | ```
208 |
209 | ## Common Activity Types
210 |
211 | ServiceNow provides several activity types that can be used when adding activities to a workflow:
212 |
213 | 1. **approval** - An approval activity that requires user action
214 | 2. **task** - A task that needs to be completed
215 | 3. **notification** - Sends a notification to users
216 | 4. **timer** - Waits for a specified amount of time
217 | 5. **condition** - Evaluates a condition and branches the workflow
218 | 6. **script** - Executes a script
219 | 7. **wait_for_condition** - Waits until a condition is met
220 | 8. **end** - Ends the workflow
221 |
222 | ## Best Practices
223 |
224 | 1. **Version Control**: Always create a new version of a workflow before making significant changes.
225 | 2. **Testing**: Test workflows in a non-production environment before deploying to production.
226 | 3. **Documentation**: Document the purpose and behavior of each workflow and activity.
227 | 4. **Error Handling**: Include error handling in your workflows to handle unexpected situations.
228 | 5. **Notifications**: Use notification activities to keep stakeholders informed about the workflow progress.
229 |
230 | ## Troubleshooting
231 |
232 | ### Common Issues
233 |
234 | 1. **Error: "No published versions found for this workflow"**
235 | - This error occurs when trying to get activities for a workflow that has no published versions.
236 | - Solution: Publish a version of the workflow before trying to get its activities.
237 |
238 | 2. **Error: "Activity type is required"**
239 | - This error occurs when trying to add an activity without specifying its type.
240 | - Solution: Provide a valid activity type when adding an activity.
241 |
242 | 3. **Error: "Cannot modify a published workflow version"**
243 | - This error occurs when trying to modify a published workflow version.
244 | - Solution: Create a new draft version of the workflow before making changes.
245 |
246 | 4. **Error: "Workflow ID is required"**
247 | - This error occurs when not providing a workflow ID for operations that require it.
248 | - Solution: Make sure to include the workflow ID in your request.
249 |
250 | ## Additional Resources
251 |
252 | - [ServiceNow Workflow Documentation](https://docs.servicenow.com/bundle/tokyo-platform-administration/page/administer/workflow-administration/concept/c_WorkflowAdministration.html)
253 | - [ServiceNow Workflow API Reference](https://developer.servicenow.com/dev.do#!/reference/api/tokyo/rest/c_WorkflowAPI)
```
--------------------------------------------------------------------------------
/docs/catalog_optimization_plan.md:
--------------------------------------------------------------------------------
```markdown
1 | # ServiceNow Catalog Optimization Plan
2 |
3 | This document outlines the plan for implementing catalog optimization features in the ServiceNow MCP integration.
4 |
5 | ## Overview
6 |
7 | 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.
8 |
9 | ## Optimization Tools
10 |
11 | ### 1. Catalog Analytics Tools
12 |
13 | #### 1.1 Get Catalog Usage Statistics
14 |
15 | **Parameters:**
16 | ```python
17 | class CatalogUsageStatsParams(BaseModel):
18 | """Parameters for getting catalog usage statistics."""
19 |
20 | time_period: str = Field("last_30_days", description="Time period for statistics (last_7_days, last_30_days, last_90_days, last_year)")
21 | category_id: Optional[str] = Field(None, description="Filter by category ID")
22 | include_inactive: bool = Field(False, description="Whether to include inactive items")
23 | ```
24 |
25 | **Returns:**
26 | - Most ordered items
27 | - Least ordered items
28 | - Average processing time
29 | - Abandonment rate (items added to cart but not ordered)
30 | - User satisfaction ratings
31 |
32 | **Implementation:**
33 | - Query the ServiceNow API for order statistics
34 | - Aggregate data by item and category
35 | - Calculate metrics like order volume, processing time, and abandonment rate
36 |
37 | #### 1.2 Get Item Performance Metrics
38 |
39 | **Parameters:**
40 | ```python
41 | class ItemPerformanceParams(BaseModel):
42 | """Parameters for getting performance metrics for a catalog item."""
43 |
44 | item_id: str = Field(..., description="Catalog item ID")
45 | time_period: str = Field("last_30_days", description="Time period for metrics")
46 | ```
47 |
48 | **Returns:**
49 | - Order volume over time
50 | - Average fulfillment time
51 | - Approval rates
52 | - Rejection reasons
53 | - User ratings and feedback
54 |
55 | **Implementation:**
56 | - Query the ServiceNow API for item-specific metrics
57 | - Calculate performance indicators
58 | - Identify trends over the specified time period
59 |
60 | ### 2. Catalog Management Tools
61 |
62 | #### 2.1 Update Catalog Item
63 |
64 | **Parameters:**
65 | ```python
66 | class UpdateCatalogItemParams(BaseModel):
67 | """Parameters for updating a catalog item."""
68 |
69 | item_id: str = Field(..., description="Catalog item ID to update")
70 | name: Optional[str] = Field(None, description="New name for the item")
71 | short_description: Optional[str] = Field(None, description="New short description")
72 | description: Optional[str] = Field(None, description="New detailed description")
73 | category: Optional[str] = Field(None, description="New category ID")
74 | price: Optional[str] = Field(None, description="New price")
75 | active: Optional[bool] = Field(None, description="Whether the item is active")
76 | order: Optional[int] = Field(None, description="Display order in the category")
77 | ```
78 |
79 | **Implementation:**
80 | - Build a PATCH request to update the catalog item
81 | - Only include fields that are provided in the parameters
82 | - Return the updated item details
83 |
84 | #### 2.2 Update Catalog Category
85 |
86 | **Parameters:**
87 | ```python
88 | class UpdateCatalogCategoryParams(BaseModel):
89 | """Parameters for updating a catalog category."""
90 |
91 | category_id: str = Field(..., description="Category ID to update")
92 | title: Optional[str] = Field(None, description="New title for the category")
93 | description: Optional[str] = Field(None, description="New description")
94 | parent: Optional[str] = Field(None, description="New parent category ID")
95 | active: Optional[bool] = Field(None, description="Whether the category is active")
96 | order: Optional[int] = Field(None, description="Display order")
97 | ```
98 |
99 | **Implementation:**
100 | - Build a PATCH request to update the catalog category
101 | - Only include fields that are provided in the parameters
102 | - Return the updated category details
103 |
104 | #### 2.3 Update Item Variable
105 |
106 | **Parameters:**
107 | ```python
108 | class UpdateItemVariableParams(BaseModel):
109 | """Parameters for updating a catalog item variable."""
110 |
111 | variable_id: str = Field(..., description="Variable ID to update")
112 | label: Optional[str] = Field(None, description="New label for the variable")
113 | help_text: Optional[str] = Field(None, description="New help text")
114 | default_value: Optional[str] = Field(None, description="New default value")
115 | mandatory: Optional[bool] = Field(None, description="Whether the variable is mandatory")
116 | order: Optional[int] = Field(None, description="Display order")
117 | ```
118 |
119 | **Implementation:**
120 | - Build a PATCH request to update the catalog item variable
121 | - Only include fields that are provided in the parameters
122 | - Return the updated variable details
123 |
124 | ### 3. Catalog Optimization Tools
125 |
126 | #### 3.1 Get Optimization Recommendations
127 |
128 | **Parameters:**
129 | ```python
130 | class OptimizationRecommendationsParams(BaseModel):
131 | """Parameters for getting catalog optimization recommendations."""
132 |
133 | category_id: Optional[str] = Field(None, description="Filter by category ID")
134 | recommendation_types: List[str] = Field(
135 | ["inactive_items", "low_usage", "high_abandonment", "slow_fulfillment", "description_quality"],
136 | description="Types of recommendations to include"
137 | )
138 | ```
139 |
140 | **Returns:**
141 | - Inactive items that could be retired
142 | - Items with low usage that might need promotion
143 | - Items with high abandonment rates that might need simplification
144 | - Items with slow fulfillment that need process improvements
145 | - Items with poor description quality
146 |
147 | **Implementation:**
148 | - Query the ServiceNow API for various metrics
149 | - Apply analysis algorithms to identify optimization opportunities
150 | - Generate recommendations based on the analysis
151 |
152 | #### 3.2 Get Catalog Structure Analysis
153 |
154 | **Parameters:**
155 | ```python
156 | class CatalogStructureAnalysisParams(BaseModel):
157 | """Parameters for analyzing catalog structure."""
158 |
159 | include_inactive: bool = Field(False, description="Whether to include inactive categories and items")
160 | ```
161 |
162 | **Returns:**
163 | - Categories with too many or too few items
164 | - Deeply nested categories that might be hard to navigate
165 | - Inconsistent naming patterns
166 | - Duplicate or similar items across categories
167 |
168 | **Implementation:**
169 | - Query the ServiceNow API for the catalog structure
170 | - Analyze the structure for usability issues
171 | - Generate recommendations for improving the structure
172 |
173 | ## Implementation Plan
174 |
175 | ### Phase 1: Analytics Tools
176 | 1. Implement `get_catalog_usage_stats`
177 | 2. Implement `get_item_performance`
178 | 3. Create tests for analytics tools
179 |
180 | ### Phase 2: Management Tools
181 | 1. Implement `update_catalog_item`
182 | 2. Implement `update_catalog_category`
183 | 3. Implement `update_item_variable`
184 | 4. Create tests for management tools
185 |
186 | ### Phase 3: Optimization Tools
187 | 1. Implement `get_optimization_recommendations`
188 | 2. Implement `get_catalog_structure_analysis`
189 | 3. Create tests for optimization tools
190 |
191 | ### Phase 4: Integration
192 | 1. Register all tools with the MCP server
193 | 2. Create example scripts for optimization workflows
194 | 3. Update documentation
195 |
196 | ## Example Usage
197 |
198 | ### Example 1: Analyzing Catalog Usage
199 |
200 | ```python
201 | # Get catalog usage statistics
202 | params = CatalogUsageStatsParams(time_period="last_90_days")
203 | result = get_catalog_usage_stats(config, auth_manager, params)
204 |
205 | # Print the most ordered items
206 | print("Most ordered items:")
207 | for item in result["most_ordered_items"]:
208 | print(f"- {item['name']}: {item['order_count']} orders")
209 |
210 | # Print items with high abandonment rates
211 | print("\nItems with high abandonment rates:")
212 | for item in result["high_abandonment_items"]:
213 | print(f"- {item['name']}: {item['abandonment_rate']}% abandonment rate")
214 | ```
215 |
216 | ### Example 2: Getting Optimization Recommendations
217 |
218 | ```python
219 | # Get optimization recommendations
220 | params = OptimizationRecommendationsParams(
221 | recommendation_types=["inactive_items", "low_usage", "description_quality"]
222 | )
223 | result = get_optimization_recommendations(config, auth_manager, params)
224 |
225 | # Print the recommendations
226 | for rec in result["recommendations"]:
227 | print(f"\n{rec['title']}")
228 | print(f"Impact: {rec['impact']}, Effort: {rec['effort']}")
229 | print(f"{rec['description']}")
230 | print(f"Recommended Action: {rec['action']}")
231 | print(f"Affected Items: {len(rec['items'])}")
232 | for item in rec['items'][:3]:
233 | print(f"- {item['name']}: {item['short_description']}")
234 | ```
235 |
236 | ### Example 3: Updating a Catalog Item
237 |
238 | ```python
239 | # Update a catalog item
240 | params = UpdateCatalogItemParams(
241 | item_id="sys_id_of_item",
242 | short_description="Updated description that is more clear and informative",
243 | price="99.99"
244 | )
245 | result = update_catalog_item(config, auth_manager, params)
246 |
247 | if result["success"]:
248 | print(f"Successfully updated item: {result['data']['name']}")
249 | else:
250 | print(f"Error: {result['message']}")
251 | ```
252 |
253 | ## Considerations
254 |
255 | 1. **Data Access**: These tools require access to ServiceNow reporting and analytics data, which might require additional permissions.
256 |
257 | 2. **Performance**: Some of these analyses could be resource-intensive, especially for large catalogs.
258 |
259 | 3. **Custom Metrics**: ServiceNow instances often have custom metrics and KPIs for catalog performance, which would need to be considered.
260 |
261 | 4. **Change Management**: Any changes to the catalog should follow proper change management processes.
262 |
263 | 5. **User Feedback**: Incorporating user feedback data would make the optimization recommendations more valuable.
```
--------------------------------------------------------------------------------
/tests/test_script_include_resources.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the script include resources.
3 |
4 | This module contains tests for the script include resources in the ServiceNow MCP server.
5 | """
6 |
7 | import json
8 | import unittest
9 | import requests
10 | from unittest.mock import MagicMock, patch
11 |
12 | from servicenow_mcp.auth.auth_manager import AuthManager
13 | from servicenow_mcp.resources.script_includes import ScriptIncludeListParams, ScriptIncludeResource
14 | from servicenow_mcp.utils.config import ServerConfig, AuthConfig, AuthType, BasicAuthConfig
15 |
16 |
17 | class TestScriptIncludeResource(unittest.IsolatedAsyncioTestCase):
18 | """Tests for the script include resource."""
19 |
20 | def setUp(self):
21 | """Set up test fixtures."""
22 | auth_config = AuthConfig(
23 | type=AuthType.BASIC,
24 | basic=BasicAuthConfig(
25 | username="test_user",
26 | password="test_password"
27 | )
28 | )
29 | self.server_config = ServerConfig(
30 | instance_url="https://test.service-now.com",
31 | auth=auth_config,
32 | )
33 | self.auth_manager = MagicMock(spec=AuthManager)
34 | self.auth_manager.get_headers.return_value = {"Authorization": "Bearer test"}
35 | self.script_include_resource = ScriptIncludeResource(self.server_config, self.auth_manager)
36 |
37 | @patch("servicenow_mcp.resources.script_includes.requests.get")
38 | async def test_list_script_includes(self, mock_get):
39 | """Test listing script includes."""
40 | # Mock response
41 | mock_response = MagicMock()
42 | mock_response.text = json.dumps({
43 | "result": [
44 | {
45 | "sys_id": "123",
46 | "name": "TestScriptInclude",
47 | "script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
48 | "description": "Test Script Include",
49 | "api_name": "global.TestScriptInclude",
50 | "client_callable": "true",
51 | "active": "true",
52 | "access": "public",
53 | "sys_created_on": "2023-01-01 00:00:00",
54 | "sys_updated_on": "2023-01-02 00:00:00",
55 | "sys_created_by": {"display_value": "admin"},
56 | "sys_updated_by": {"display_value": "admin"}
57 | }
58 | ]
59 | })
60 | mock_response.status_code = 200
61 | mock_get.return_value = mock_response
62 |
63 | # Call the method
64 | params = ScriptIncludeListParams(
65 | limit=10,
66 | offset=0,
67 | active=True,
68 | client_callable=True,
69 | query="Test"
70 | )
71 | result = await self.script_include_resource.list_script_includes(params)
72 | result_json = json.loads(result)
73 |
74 | # Verify the result
75 | self.assertIn("result", result_json)
76 | self.assertEqual(1, len(result_json["result"]))
77 | self.assertEqual("123", result_json["result"][0]["sys_id"])
78 | self.assertEqual("TestScriptInclude", result_json["result"][0]["name"])
79 | self.assertEqual("true", result_json["result"][0]["client_callable"])
80 | self.assertEqual("true", result_json["result"][0]["active"])
81 |
82 | # Verify the request
83 | mock_get.assert_called_once()
84 | args, kwargs = mock_get.call_args
85 | self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
86 | self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
87 | self.assertEqual(10, kwargs["params"]["sysparm_limit"])
88 | self.assertEqual(0, kwargs["params"]["sysparm_offset"])
89 | self.assertEqual("active=true^client_callable=true^nameLIKETest", kwargs["params"]["sysparm_query"])
90 |
91 | @patch("servicenow_mcp.resources.script_includes.requests.get")
92 | async def test_get_script_include(self, mock_get):
93 | """Test getting a script include."""
94 | # Mock response
95 | mock_response = MagicMock()
96 | mock_response.text = json.dumps({
97 | "result": {
98 | "sys_id": "123",
99 | "name": "TestScriptInclude",
100 | "script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
101 | "description": "Test Script Include",
102 | "api_name": "global.TestScriptInclude",
103 | "client_callable": "true",
104 | "active": "true",
105 | "access": "public",
106 | "sys_created_on": "2023-01-01 00:00:00",
107 | "sys_updated_on": "2023-01-02 00:00:00",
108 | "sys_created_by": {"display_value": "admin"},
109 | "sys_updated_by": {"display_value": "admin"}
110 | }
111 | })
112 | mock_response.status_code = 200
113 | mock_get.return_value = mock_response
114 |
115 | # Call the method
116 | result = await self.script_include_resource.get_script_include("123")
117 | result_json = json.loads(result)
118 |
119 | # Verify the result
120 | self.assertIn("result", result_json)
121 | self.assertEqual("123", result_json["result"]["sys_id"])
122 | self.assertEqual("TestScriptInclude", result_json["result"]["name"])
123 | self.assertEqual("true", result_json["result"]["client_callable"])
124 | self.assertEqual("true", result_json["result"]["active"])
125 |
126 | # Verify the request
127 | mock_get.assert_called_once()
128 | args, kwargs = mock_get.call_args
129 | self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include", args[0])
130 | self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
131 | self.assertEqual("name=123", kwargs["params"]["sysparm_query"])
132 |
133 | @patch("servicenow_mcp.resources.script_includes.requests.get")
134 | async def test_get_script_include_by_sys_id(self, mock_get):
135 | """Test getting a script include by sys_id."""
136 | # Mock response
137 | mock_response = MagicMock()
138 | mock_response.text = json.dumps({
139 | "result": {
140 | "sys_id": "123",
141 | "name": "TestScriptInclude",
142 | "script": "var TestScriptInclude = Class.create();\nTestScriptInclude.prototype = {\n initialize: function() {\n },\n\n type: 'TestScriptInclude'\n};",
143 | "description": "Test Script Include",
144 | "api_name": "global.TestScriptInclude",
145 | "client_callable": "true",
146 | "active": "true",
147 | "access": "public",
148 | "sys_created_on": "2023-01-01 00:00:00",
149 | "sys_updated_on": "2023-01-02 00:00:00",
150 | "sys_created_by": {"display_value": "admin"},
151 | "sys_updated_by": {"display_value": "admin"}
152 | }
153 | })
154 | mock_response.status_code = 200
155 | mock_get.return_value = mock_response
156 |
157 | # Call the method
158 | result = await self.script_include_resource.get_script_include("sys_id:123")
159 | result_json = json.loads(result)
160 |
161 | # Verify the result
162 | self.assertIn("result", result_json)
163 | self.assertEqual("123", result_json["result"]["sys_id"])
164 | self.assertEqual("TestScriptInclude", result_json["result"]["name"])
165 |
166 | # Verify the request
167 | mock_get.assert_called_once()
168 | args, kwargs = mock_get.call_args
169 | self.assertEqual(f"{self.server_config.instance_url}/api/now/table/sys_script_include/123", args[0])
170 | self.assertEqual({"Authorization": "Bearer test"}, kwargs["headers"])
171 |
172 | @patch("servicenow_mcp.resources.script_includes.requests.get")
173 | async def test_list_script_includes_error(self, mock_get):
174 | """Test listing script includes with an error."""
175 | # Mock response
176 | mock_get.side_effect = requests.RequestException("Test error")
177 |
178 | # Call the method
179 | params = ScriptIncludeListParams()
180 | result = await self.script_include_resource.list_script_includes(params)
181 | result_json = json.loads(result)
182 |
183 | # Verify the result
184 | self.assertIn("error", result_json)
185 | self.assertIn("Error listing script includes", result_json["error"])
186 |
187 | @patch("servicenow_mcp.resources.script_includes.requests.get")
188 | async def test_get_script_include_error(self, mock_get):
189 | """Test getting a script include with an error."""
190 | # Mock response
191 | mock_get.side_effect = requests.RequestException("Test error")
192 |
193 | # Call the method
194 | result = await self.script_include_resource.get_script_include("123")
195 | result_json = json.loads(result)
196 |
197 | # Verify the result
198 | self.assertIn("error", result_json)
199 | self.assertIn("Error getting script include", result_json["error"])
200 |
201 |
202 | class TestScriptIncludeListParams(unittest.TestCase):
203 | """Tests for the script include list parameters."""
204 |
205 | def test_script_include_list_params(self):
206 | """Test script include list parameters."""
207 | params = ScriptIncludeListParams(
208 | limit=20,
209 | offset=10,
210 | active=True,
211 | client_callable=False,
212 | query="Test"
213 | )
214 | self.assertEqual(20, params.limit)
215 | self.assertEqual(10, params.offset)
216 | self.assertTrue(params.active)
217 | self.assertFalse(params.client_callable)
218 | self.assertEqual("Test", params.query)
219 |
220 | def test_script_include_list_params_defaults(self):
221 | """Test script include list parameters defaults."""
222 | params = ScriptIncludeListParams()
223 | self.assertEqual(10, params.limit)
224 | self.assertEqual(0, params.offset)
225 | self.assertIsNone(params.active)
226 | self.assertIsNone(params.client_callable)
227 | self.assertIsNone(params.query)
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/cli.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Command-line interface for the ServiceNow MCP server.
3 | """
4 |
5 | import argparse
6 | import logging
7 | import os
8 | import sys
9 |
10 | import anyio
11 | from dotenv import load_dotenv
12 | from mcp.server.stdio import stdio_server
13 |
14 | from servicenow_mcp.server import ServiceNowMCP
15 | from servicenow_mcp.utils.config import (
16 | ApiKeyConfig,
17 | AuthConfig,
18 | AuthType,
19 | BasicAuthConfig,
20 | OAuthConfig,
21 | ServerConfig,
22 | )
23 |
24 | # Configure logging
25 | logging.basicConfig(
26 | level=logging.INFO,
27 | format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
28 | )
29 | logger = logging.getLogger(__name__)
30 |
31 |
32 | def parse_args():
33 | """Parse command-line arguments."""
34 | parser = argparse.ArgumentParser(description="ServiceNow MCP Server")
35 |
36 | # Server configuration
37 | parser.add_argument(
38 | "--instance-url",
39 | help="ServiceNow instance URL (e.g., https://instance.service-now.com)",
40 | default=os.environ.get("SERVICENOW_INSTANCE_URL"),
41 | )
42 | parser.add_argument(
43 | "--debug",
44 | action="store_true",
45 | help="Enable debug mode",
46 | default=os.environ.get("SERVICENOW_DEBUG", "false").lower() == "true",
47 | )
48 | parser.add_argument(
49 | "--timeout",
50 | type=int,
51 | help="Request timeout in seconds",
52 | default=int(os.environ.get("SERVICENOW_TIMEOUT", "30")),
53 | )
54 |
55 | # Authentication
56 | auth_group = parser.add_argument_group("Authentication")
57 | auth_group.add_argument(
58 | "--auth-type",
59 | choices=["basic", "oauth", "api_key"],
60 | help="Authentication type",
61 | default=os.environ.get("SERVICENOW_AUTH_TYPE", "basic"),
62 | )
63 |
64 | # Basic auth
65 | basic_group = parser.add_argument_group("Basic Authentication")
66 | basic_group.add_argument(
67 | "--username",
68 | help="ServiceNow username",
69 | default=os.environ.get("SERVICENOW_USERNAME"),
70 | )
71 | basic_group.add_argument(
72 | "--password",
73 | help="ServiceNow password",
74 | default=os.environ.get("SERVICENOW_PASSWORD"),
75 | )
76 |
77 | # OAuth
78 | oauth_group = parser.add_argument_group("OAuth Authentication")
79 | oauth_group.add_argument(
80 | "--client-id",
81 | help="OAuth client ID",
82 | default=os.environ.get("SERVICENOW_CLIENT_ID"),
83 | )
84 | oauth_group.add_argument(
85 | "--client-secret",
86 | help="OAuth client secret",
87 | default=os.environ.get("SERVICENOW_CLIENT_SECRET"),
88 | )
89 | oauth_group.add_argument(
90 | "--token-url",
91 | help="OAuth token URL",
92 | default=os.environ.get("SERVICENOW_TOKEN_URL"),
93 | )
94 |
95 | # API Key
96 | api_key_group = parser.add_argument_group("API Key Authentication")
97 | api_key_group.add_argument(
98 | "--api-key",
99 | help="ServiceNow API key",
100 | default=os.environ.get("SERVICENOW_API_KEY"),
101 | )
102 | api_key_group.add_argument(
103 | "--api-key-header",
104 | help="API key header name",
105 | default=os.environ.get("SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key"),
106 | )
107 |
108 | # Script execution API resource path
109 | script_execution_group = parser.add_argument_group("Script Execution API")
110 | script_execution_group.add_argument(
111 | "--script-execution-api-resource-path",
112 | help="Script execution API resource path",
113 | default=os.environ.get("SCRIPT_EXECUTION_API_RESOURCE_PATH"),
114 | )
115 |
116 | return parser.parse_args()
117 |
118 |
119 | def create_config(args) -> ServerConfig:
120 | """
121 | Create server configuration from command-line arguments.
122 |
123 | Args:
124 | args: Command-line arguments.
125 |
126 | Returns:
127 | ServerConfig: Server configuration.
128 |
129 | Raises:
130 | ValueError: If required configuration is missing.
131 | """
132 | # NOTE: This assumes the ServerConfig model takes instance_url, auth, debug, timeout etc.
133 | # The ServiceNowMCP class now expects a ServerConfig object matching this.
134 |
135 | # Instance URL validation
136 | instance_url = args.instance_url
137 | if not instance_url:
138 | # Attempt to load from .env if not provided via args/env vars directly in parse_args
139 | instance_url = os.getenv("SERVICENOW_INSTANCE_URL")
140 | if not instance_url:
141 | raise ValueError(
142 | "ServiceNow instance URL is required (--instance-url or SERVICENOW_INSTANCE_URL env var)"
143 | )
144 |
145 | # Create authentication configuration based on args
146 | auth_type = AuthType(args.auth_type.lower())
147 | # This will hold the final AuthConfig instance for ServerConfig
148 | final_auth_config: AuthConfig
149 |
150 | if auth_type == AuthType.BASIC:
151 | username = args.username or os.getenv("SERVICENOW_USERNAME")
152 | password = args.password or os.getenv("SERVICENOW_PASSWORD") # Get password from arg or env
153 | if not username or not password:
154 | raise ValueError(
155 | "Username and password are required for basic authentication (--username/SERVICENOW_USERNAME, --password/SERVICENOW_PASSWORD)"
156 | )
157 | # Create the specific config (without instance_url)
158 | basic_cfg = BasicAuthConfig(
159 | username=username,
160 | password=password,
161 | )
162 | # Create the main AuthConfig wrapper
163 | final_auth_config = AuthConfig(type=auth_type, basic=basic_cfg)
164 |
165 | elif auth_type == AuthType.OAUTH:
166 | # Simplified - assuming password grant for now based on previous args
167 | client_id = args.client_id or os.getenv("SERVICENOW_CLIENT_ID")
168 | client_secret = args.client_secret or os.getenv("SERVICENOW_CLIENT_SECRET")
169 | username = args.username or os.getenv("SERVICENOW_USERNAME") # Needed for password grant
170 | password = args.password or os.getenv("SERVICENOW_PASSWORD") # Needed for password grant
171 | token_url = args.token_url or os.getenv("SERVICENOW_TOKEN_URL")
172 |
173 | if not client_id or not client_secret or not username or not password:
174 | raise ValueError(
175 | "Client ID, client secret, username, and password are required for OAuth password grant"
176 | " (--client-id/SERVICENOW_CLIENT_ID, etc.)"
177 | )
178 | if not token_url:
179 | # Attempt to construct default if not provided
180 | token_url = f"{instance_url}/oauth_token.do"
181 | logger.warning(f"OAuth token URL not provided, defaulting to: {token_url}")
182 |
183 | # Create the specific config (without instance_url)
184 | oauth_cfg = OAuthConfig(
185 | client_id=client_id,
186 | client_secret=client_secret,
187 | username=username,
188 | password=password,
189 | token_url=token_url,
190 | )
191 | # Create the main AuthConfig wrapper
192 | final_auth_config = AuthConfig(type=auth_type, oauth=oauth_cfg)
193 |
194 | elif auth_type == AuthType.API_KEY:
195 | api_key = args.api_key or os.getenv("SERVICENOW_API_KEY")
196 | api_key_header = args.api_key_header or os.getenv(
197 | "SERVICENOW_API_KEY_HEADER", "X-ServiceNow-API-Key"
198 | )
199 | if not api_key:
200 | raise ValueError(
201 | "API key is required for API key authentication (--api-key or SERVICENOW_API_KEY)"
202 | )
203 | # Create the specific config (without instance_url)
204 | api_key_cfg = ApiKeyConfig(
205 | api_key=api_key,
206 | header_name=api_key_header,
207 | )
208 | # Create the main AuthConfig wrapper
209 | final_auth_config = AuthConfig(type=auth_type, api_key=api_key_cfg)
210 | else:
211 | # Should not happen if choices are enforced by argparse
212 | raise ValueError(f"Unsupported authentication type: {args.auth_type}")
213 |
214 | # Script execution path
215 | script_execution_api_resource_path = args.script_execution_api_resource_path or os.getenv(
216 | "SCRIPT_EXECUTION_API_RESOURCE_PATH"
217 | )
218 | if not script_execution_api_resource_path:
219 | logger.warning(
220 | "Script execution API resource path not set (--script-execution-api-resource-path or SCRIPT_EXECUTION_API_RESOURCE_PATH). ExecuteScriptInclude tool may fail."
221 | )
222 |
223 | # Create the final ServerConfig
224 | # Ensure ServerConfig model expects 'auth' as a nested object
225 | return ServerConfig(
226 | instance_url=instance_url, # Add instance_url directly here
227 | auth=final_auth_config, # Pass the correctly structured AuthConfig instance
228 | # Include other server config fields if they exist on ServerConfig model
229 | debug=args.debug,
230 | timeout=args.timeout,
231 | script_execution_api_resource_path=script_execution_api_resource_path,
232 | )
233 |
234 |
235 | async def arun_server(server_instance):
236 | """Runs the given MCP server instance using stdio transport."""
237 | logger.info("Starting server with stdio transport...")
238 | async with stdio_server() as streams:
239 | # Get initialization options from the low-level server
240 | init_options = server_instance.create_initialization_options()
241 | await server_instance.run(streams[0], streams[1], init_options)
242 | logger.info("Stdio server finished.")
243 |
244 |
245 | def main():
246 | """Main entry point for the CLI."""
247 | # Load environment variables from .env file
248 | load_dotenv()
249 |
250 | try:
251 | # Parse command-line arguments
252 | args = parse_args()
253 |
254 | # Configure logging level based on debug flag
255 | if args.debug:
256 | logging.getLogger().setLevel(logging.DEBUG)
257 | logger.info("Debug logging enabled.")
258 | else:
259 | logging.getLogger().setLevel(logging.INFO)
260 |
261 | # Create server configuration
262 | config = create_config(args)
263 | # Log the instance URL being used (mask sensitive parts of config if needed)
264 | logger.info(f"Initializing ServiceNow MCP server for instance: {config.instance_url}")
265 |
266 | # Create server controller instance
267 | mcp_controller = ServiceNowMCP(config)
268 |
269 | # Get the low-level server instance to run
270 | server_to_run = mcp_controller.start()
271 |
272 | # Run the server using anyio and the stdio transport
273 | anyio.run(arun_server, server_to_run)
274 |
275 | except ValueError as e:
276 | logger.error(f"Configuration or runtime error: {e}")
277 | sys.exit(1)
278 |
279 | except Exception as e:
280 | logger.exception(f"Unexpected error starting or running server: {e}")
281 | sys.exit(1)
282 |
283 |
284 | if __name__ == "__main__":
285 | main()
286 |
```
--------------------------------------------------------------------------------
/docs/user_management.md:
--------------------------------------------------------------------------------
```markdown
1 | # User Management in ServiceNow MCP
2 |
3 | This document provides detailed information about the User Management tools available in the ServiceNow MCP server.
4 |
5 | ## Overview
6 |
7 | 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.
8 |
9 | ## Available Tools
10 |
11 | ### User Management
12 |
13 | 1. **create_user** - Create a new user in ServiceNow
14 | 2. **update_user** - Update an existing user in ServiceNow
15 | 3. **get_user** - Get a specific user by ID, username, or email
16 | 4. **list_users** - List users with filtering options
17 |
18 | ### Group Management
19 |
20 | 5. **create_group** - Create a new group in ServiceNow
21 | 6. **update_group** - Update an existing group in ServiceNow
22 | 7. **add_group_members** - Add members to a group in ServiceNow
23 | 8. **remove_group_members** - Remove members from a group in ServiceNow
24 | 9. **list_groups** - List groups with filtering options
25 |
26 | ## Tool Details
27 |
28 | ### create_user
29 |
30 | Creates a new user in ServiceNow.
31 |
32 | #### Parameters
33 |
34 | | Parameter | Type | Required | Description |
35 | |-----------|------|----------|-------------|
36 | | user_name | string | Yes | Username for the user |
37 | | first_name | string | Yes | First name of the user |
38 | | last_name | string | Yes | Last name of the user |
39 | | email | string | Yes | Email address of the user |
40 | | title | string | No | Job title of the user |
41 | | department | string | No | Department the user belongs to |
42 | | manager | string | No | Manager of the user (sys_id or username) |
43 | | roles | array | No | Roles to assign to the user |
44 | | phone | string | No | Phone number of the user |
45 | | mobile_phone | string | No | Mobile phone number of the user |
46 | | location | string | No | Location of the user |
47 | | password | string | No | Password for the user account |
48 | | active | boolean | No | Whether the user account is active (default: true) |
49 |
50 | #### Example
51 |
52 | ```python
53 | # Create a new user in the Radiology department
54 | result = create_user({
55 | "user_name": "alice.radiology",
56 | "first_name": "Alice",
57 | "last_name": "Radiology",
58 | "email": "[email protected]",
59 | "title": "Doctor",
60 | "department": "Radiology",
61 | "roles": ["user"]
62 | })
63 | ```
64 |
65 | ### update_user
66 |
67 | Updates an existing user in ServiceNow.
68 |
69 | #### Parameters
70 |
71 | | Parameter | Type | Required | Description |
72 | |-----------|------|----------|-------------|
73 | | user_id | string | Yes | User ID or sys_id to update |
74 | | user_name | string | No | Username for the user |
75 | | first_name | string | No | First name of the user |
76 | | last_name | string | No | Last name of the user |
77 | | email | string | No | Email address of the user |
78 | | title | string | No | Job title of the user |
79 | | department | string | No | Department the user belongs to |
80 | | manager | string | No | Manager of the user (sys_id or username) |
81 | | roles | array | No | Roles to assign to the user |
82 | | phone | string | No | Phone number of the user |
83 | | mobile_phone | string | No | Mobile phone number of the user |
84 | | location | string | No | Location of the user |
85 | | password | string | No | Password for the user account |
86 | | active | boolean | No | Whether the user account is active |
87 |
88 | #### Example
89 |
90 | ```python
91 | # Update a user to set their manager
92 | result = update_user({
93 | "user_id": "user123",
94 | "manager": "user456",
95 | "title": "Senior Doctor"
96 | })
97 | ```
98 |
99 | ### get_user
100 |
101 | Gets a specific user from ServiceNow.
102 |
103 | #### Parameters
104 |
105 | | Parameter | Type | Required | Description |
106 | |-----------|------|----------|-------------|
107 | | user_id | string | No | User ID or sys_id |
108 | | user_name | string | No | Username of the user |
109 | | email | string | No | Email address of the user |
110 |
111 | **Note**: At least one of the parameters must be provided.
112 |
113 | #### Example
114 |
115 | ```python
116 | # Get a user by username
117 | result = get_user({
118 | "user_name": "alice.radiology"
119 | })
120 | ```
121 |
122 | ### list_users
123 |
124 | Lists users from ServiceNow with filtering options.
125 |
126 | #### Parameters
127 |
128 | | Parameter | Type | Required | Description |
129 | |-----------|------|----------|-------------|
130 | | limit | integer | No | Maximum number of users to return (default: 10) |
131 | | offset | integer | No | Offset for pagination (default: 0) |
132 | | active | boolean | No | Filter by active status |
133 | | department | string | No | Filter by department |
134 | | query | string | No | Search query for users |
135 |
136 | #### Example
137 |
138 | ```python
139 | # List users in the Radiology department
140 | result = list_users({
141 | "department": "Radiology",
142 | "active": true,
143 | "limit": 20
144 | })
145 | ```
146 |
147 | ### create_group
148 |
149 | Creates a new group in ServiceNow.
150 |
151 | #### Parameters
152 |
153 | | Parameter | Type | Required | Description |
154 | |-----------|------|----------|-------------|
155 | | name | string | Yes | Name of the group |
156 | | description | string | No | Description of the group |
157 | | manager | string | No | Manager of the group (sys_id or username) |
158 | | parent | string | No | Parent group (sys_id or name) |
159 | | type | string | No | Type of the group |
160 | | email | string | No | Email address for the group |
161 | | members | array | No | List of user sys_ids or usernames to add as members |
162 | | active | boolean | No | Whether the group is active (default: true) |
163 |
164 | #### Example
165 |
166 | ```python
167 | # Create a new group for Biomedical Engineering
168 | result = create_group({
169 | "name": "Biomedical Engineering",
170 | "description": "Group for biomedical engineering staff",
171 | "manager": "user456",
172 | "members": ["admin", "alice.radiology"]
173 | })
174 | ```
175 |
176 | ### update_group
177 |
178 | Updates an existing group in ServiceNow.
179 |
180 | #### Parameters
181 |
182 | | Parameter | Type | Required | Description |
183 | |-----------|------|----------|-------------|
184 | | group_id | string | Yes | Group ID or sys_id to update |
185 | | name | string | No | Name of the group |
186 | | description | string | No | Description of the group |
187 | | manager | string | No | Manager of the group (sys_id or username) |
188 | | parent | string | No | Parent group (sys_id or name) |
189 | | type | string | No | Type of the group |
190 | | email | string | No | Email address for the group |
191 | | active | boolean | No | Whether the group is active |
192 |
193 | #### Example
194 |
195 | ```python
196 | # Update a group to change its manager
197 | result = update_group({
198 | "group_id": "group123",
199 | "description": "Updated description for biomedical engineering group",
200 | "manager": "user789"
201 | })
202 | ```
203 |
204 | ### add_group_members
205 |
206 | Adds members to a group in ServiceNow.
207 |
208 | #### Parameters
209 |
210 | | Parameter | Type | Required | Description |
211 | |-----------|------|----------|-------------|
212 | | group_id | string | Yes | Group ID or sys_id |
213 | | members | array | Yes | List of user sys_ids or usernames to add as members |
214 |
215 | #### Example
216 |
217 | ```python
218 | # Add members to the Biomedical Engineering group
219 | result = add_group_members({
220 | "group_id": "group123",
221 | "members": ["bob.chiefradiology", "admin"]
222 | })
223 | ```
224 |
225 | ### remove_group_members
226 |
227 | Removes members from a group in ServiceNow.
228 |
229 | #### Parameters
230 |
231 | | Parameter | Type | Required | Description |
232 | |-----------|------|----------|-------------|
233 | | group_id | string | Yes | Group ID or sys_id |
234 | | members | array | Yes | List of user sys_ids or usernames to remove as members |
235 |
236 | #### Example
237 |
238 | ```python
239 | # Remove a member from the Biomedical Engineering group
240 | result = remove_group_members({
241 | "group_id": "group123",
242 | "members": ["alice.radiology"]
243 | })
244 | ```
245 |
246 | ### list_groups
247 |
248 | Lists groups from ServiceNow with filtering options.
249 |
250 | #### Parameters
251 |
252 | | Parameter | Type | Required | Description |
253 | |-----------|------|----------|-------------|
254 | | limit | integer | No | Maximum number of groups to return (default: 10) |
255 | | offset | integer | No | Offset for pagination (default: 0) |
256 | | active | boolean | No | Filter by active status |
257 | | type | string | No | Filter by group type |
258 | | query | string | No | Case-insensitive search term that matches against group name or description fields. Uses ServiceNow's LIKE operator for partial matching. |
259 |
260 | #### Example
261 |
262 | ```python
263 | # List active IT-type groups
264 | result = list_groups({
265 | "active": true,
266 | "type": "it",
267 | "query": "support",
268 | "limit": 20
269 | })
270 | ```
271 |
272 | ## Common Scenarios
273 |
274 | ### Creating Test Users and Groups for Approval Workflows
275 |
276 | To set up test users and groups for an approval workflow:
277 |
278 | 1. Create department head user:
279 |
280 | ```python
281 | bob_result = create_user({
282 | "user_name": "bob.chiefradiology",
283 | "first_name": "Bob",
284 | "last_name": "ChiefRadiology",
285 | "email": "[email protected]",
286 | "title": "Chief of Radiology",
287 | "department": "Radiology",
288 | "roles": ["itil", "admin"] # assign ITIL role for approvals
289 | })
290 | ```
291 |
292 | 2. Create staff user with department head as manager:
293 |
294 | ```python
295 | alice_result = create_user({
296 | "user_name": "alice.radiology",
297 | "first_name": "Alice",
298 | "last_name": "Radiology",
299 | "email": "[email protected]",
300 | "title": "Doctor",
301 | "department": "Radiology",
302 | "manager": bob_result.user_id # Set Bob as Alice's manager
303 | })
304 | ```
305 |
306 | 3. Create assignment group for fulfillment:
307 |
308 | ```python
309 | group_result = create_group({
310 | "name": "Biomedical Engineering",
311 | "description": "Group for biomedical engineering staff",
312 | "members": ["admin"] # Add administrator as a member
313 | })
314 | ```
315 |
316 | ### Finding Users in a Department
317 |
318 | To find all users in a specific department:
319 |
320 | ```python
321 | users = list_users({
322 | "department": "Radiology",
323 | "limit": 50
324 | })
325 | ```
326 |
327 | ### Setting up Role-Based Access Control
328 |
329 | To assign specific roles to users:
330 |
331 | ```python
332 | # Assign ITIL role to a user so they can approve changes
333 | update_user({
334 | "user_id": "user123",
335 | "roles": ["itil"]
336 | })
337 | ```
338 |
339 | ## Troubleshooting
340 |
341 | ### Common Errors
342 |
343 | 1. **User already exists**
344 | - This error occurs when trying to create a user with a username that already exists
345 | - Solution: Use a different username or update the existing user instead
346 |
347 | 2. **User not found**
348 | - This error occurs when trying to update, get, or add a user that doesn't exist
349 | - Solution: Verify the user ID, username, or email is correct
350 |
351 | 3. **Role not found**
352 | - This error occurs when trying to assign a role that doesn't exist
353 | - Solution: Check the role name and make sure it exists in the ServiceNow instance
354 |
355 | 4. **Group not found**
356 | - This error occurs when trying to update or add members to a group that doesn't exist
357 | - Solution: Verify the group ID is correct
358 |
359 | ## Best Practices
360 |
361 | 1. **Use meaningful usernames**: Create usernames that reflect the user's identity, such as "firstname.lastname"
362 |
363 | 2. **Set up proper role assignments**: Only assign the necessary roles to users to maintain security best practices
364 |
365 | 3. **Organize users into appropriate groups**: Use groups to organize users based on departments, functions, or teams
366 |
367 | 4. **Manage group memberships carefully**: Add or remove users from groups to ensure proper assignment and notification routing
368 |
369 | 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
1 | """
2 | Catalog Item Variables tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing variables (form fields) in ServiceNow catalog items.
5 | """
6 |
7 | import logging
8 | from typing import Any, Dict, List, Optional
9 |
10 | import requests
11 | from pydantic import BaseModel, Field
12 |
13 | from servicenow_mcp.auth.auth_manager import AuthManager
14 | from servicenow_mcp.utils.config import ServerConfig
15 |
16 | logger = logging.getLogger(__name__)
17 |
18 |
19 | class CreateCatalogItemVariableParams(BaseModel):
20 | """Parameters for creating a catalog item variable."""
21 |
22 | catalog_item_id: str = Field(..., description="The sys_id of the catalog item")
23 | name: str = Field(..., description="The name of the variable (internal name)")
24 | type: str = Field(..., description="The type of variable (e.g., string, integer, boolean, reference)")
25 | label: str = Field(..., description="The display label for the variable")
26 | mandatory: bool = Field(False, description="Whether the variable is required")
27 | help_text: Optional[str] = Field(None, description="Help text to display with the variable")
28 | default_value: Optional[str] = Field(None, description="Default value for the variable")
29 | description: Optional[str] = Field(None, description="Description of the variable")
30 | order: Optional[int] = Field(None, description="Display order of the variable")
31 | reference_table: Optional[str] = Field(None, description="For reference fields, the table to reference")
32 | reference_qualifier: Optional[str] = Field(None, description="For reference fields, the query to filter reference options")
33 | max_length: Optional[int] = Field(None, description="Maximum length for string fields")
34 | min: Optional[int] = Field(None, description="Minimum value for numeric fields")
35 | max: Optional[int] = Field(None, description="Maximum value for numeric fields")
36 |
37 |
38 | class CatalogItemVariableResponse(BaseModel):
39 | """Response from catalog item variable operations."""
40 |
41 | success: bool = Field(..., description="Whether the operation was successful")
42 | message: str = Field(..., description="Message describing the result")
43 | variable_id: Optional[str] = Field(None, description="The sys_id of the created/updated variable")
44 | details: Optional[Dict[str, Any]] = Field(None, description="Additional details about the variable")
45 |
46 |
47 | class ListCatalogItemVariablesParams(BaseModel):
48 | """Parameters for listing catalog item variables."""
49 |
50 | catalog_item_id: str = Field(..., description="The sys_id of the catalog item")
51 | include_details: bool = Field(True, description="Whether to include detailed information about each variable")
52 | limit: Optional[int] = Field(None, description="Maximum number of variables to return")
53 | offset: Optional[int] = Field(None, description="Offset for pagination")
54 |
55 |
56 | class ListCatalogItemVariablesResponse(BaseModel):
57 | """Response from listing catalog item variables."""
58 |
59 | success: bool = Field(..., description="Whether the operation was successful")
60 | message: str = Field(..., description="Message describing the result")
61 | variables: List[Dict[str, Any]] = Field([], description="List of variables")
62 | count: int = Field(0, description="Total number of variables found")
63 |
64 |
65 | class UpdateCatalogItemVariableParams(BaseModel):
66 | """Parameters for updating a catalog item variable."""
67 |
68 | variable_id: str = Field(..., description="The sys_id of the variable to update")
69 | label: Optional[str] = Field(None, description="The display label for the variable")
70 | mandatory: Optional[bool] = Field(None, description="Whether the variable is required")
71 | help_text: Optional[str] = Field(None, description="Help text to display with the variable")
72 | default_value: Optional[str] = Field(None, description="Default value for the variable")
73 | description: Optional[str] = Field(None, description="Description of the variable")
74 | order: Optional[int] = Field(None, description="Display order of the variable")
75 | reference_qualifier: Optional[str] = Field(None, description="For reference fields, the query to filter reference options")
76 | max_length: Optional[int] = Field(None, description="Maximum length for string fields")
77 | min: Optional[int] = Field(None, description="Minimum value for numeric fields")
78 | max: Optional[int] = Field(None, description="Maximum value for numeric fields")
79 |
80 |
81 | def create_catalog_item_variable(
82 | config: ServerConfig,
83 | auth_manager: AuthManager,
84 | params: CreateCatalogItemVariableParams,
85 | ) -> CatalogItemVariableResponse:
86 | """
87 | Create a new variable (form field) for a catalog item.
88 |
89 | Args:
90 | config: Server configuration.
91 | auth_manager: Authentication manager.
92 | params: Parameters for creating a catalog item variable.
93 |
94 | Returns:
95 | Response with information about the created variable.
96 | """
97 | api_url = f"{config.instance_url}/api/now/table/item_option_new"
98 |
99 | # Build request data
100 | data = {
101 | "cat_item": params.catalog_item_id,
102 | "name": params.name,
103 | "type": params.type,
104 | "question_text": params.label,
105 | "mandatory": str(params.mandatory).lower(), # ServiceNow expects "true"/"false" strings
106 | }
107 |
108 | if params.help_text:
109 | data["help_text"] = params.help_text
110 | if params.default_value:
111 | data["default_value"] = params.default_value
112 | if params.description:
113 | data["description"] = params.description
114 | if params.order is not None:
115 | data["order"] = params.order
116 | if params.reference_table:
117 | data["reference"] = params.reference_table
118 | if params.reference_qualifier:
119 | data["reference_qual"] = params.reference_qualifier
120 | if params.max_length:
121 | data["max_length"] = params.max_length
122 | if params.min is not None:
123 | data["min"] = params.min
124 | if params.max is not None:
125 | data["max"] = params.max
126 |
127 | # Make request
128 | try:
129 | response = requests.post(
130 | api_url,
131 | json=data,
132 | headers=auth_manager.get_headers(),
133 | timeout=config.timeout,
134 | )
135 | response.raise_for_status()
136 |
137 | result = response.json().get("result", {})
138 |
139 | return CatalogItemVariableResponse(
140 | success=True,
141 | message="Catalog item variable created successfully",
142 | variable_id=result.get("sys_id"),
143 | details=result,
144 | )
145 |
146 | except requests.RequestException as e:
147 | logger.error(f"Failed to create catalog item variable: {e}")
148 | return CatalogItemVariableResponse(
149 | success=False,
150 | message=f"Failed to create catalog item variable: {str(e)}",
151 | )
152 |
153 |
154 | def list_catalog_item_variables(
155 | config: ServerConfig,
156 | auth_manager: AuthManager,
157 | params: ListCatalogItemVariablesParams,
158 | ) -> ListCatalogItemVariablesResponse:
159 | """
160 | List all variables (form fields) for a catalog item.
161 |
162 | Args:
163 | config: Server configuration.
164 | auth_manager: Authentication manager.
165 | params: Parameters for listing catalog item variables.
166 |
167 | Returns:
168 | Response with a list of variables for the catalog item.
169 | """
170 | # Build query parameters
171 | query_params = {
172 | "sysparm_query": f"cat_item={params.catalog_item_id}^ORDERBYorder",
173 | }
174 |
175 | if params.limit:
176 | query_params["sysparm_limit"] = params.limit
177 | if params.offset:
178 | query_params["sysparm_offset"] = params.offset
179 |
180 | # Include all fields if detailed info is requested
181 | if params.include_details:
182 | query_params["sysparm_display_value"] = "true"
183 | query_params["sysparm_exclude_reference_link"] = "false"
184 | else:
185 | query_params["sysparm_fields"] = "sys_id,name,type,question_text,order,mandatory"
186 |
187 | api_url = f"{config.instance_url}/api/now/table/item_option_new"
188 |
189 | # Make request
190 | try:
191 | response = requests.get(
192 | api_url,
193 | params=query_params,
194 | headers=auth_manager.get_headers(),
195 | timeout=config.timeout,
196 | )
197 | response.raise_for_status()
198 |
199 | result = response.json().get("result", [])
200 |
201 | return ListCatalogItemVariablesResponse(
202 | success=True,
203 | message=f"Retrieved {len(result)} variables for catalog item",
204 | variables=result,
205 | count=len(result),
206 | )
207 |
208 | except requests.RequestException as e:
209 | logger.error(f"Failed to list catalog item variables: {e}")
210 | return ListCatalogItemVariablesResponse(
211 | success=False,
212 | message=f"Failed to list catalog item variables: {str(e)}",
213 | )
214 |
215 |
216 | def update_catalog_item_variable(
217 | config: ServerConfig,
218 | auth_manager: AuthManager,
219 | params: UpdateCatalogItemVariableParams,
220 | ) -> CatalogItemVariableResponse:
221 | """
222 | Update an existing variable (form field) for a catalog item.
223 |
224 | Args:
225 | config: Server configuration.
226 | auth_manager: Authentication manager.
227 | params: Parameters for updating a catalog item variable.
228 |
229 | Returns:
230 | Response with information about the updated variable.
231 | """
232 | api_url = f"{config.instance_url}/api/now/table/item_option_new/{params.variable_id}"
233 |
234 | # Build request data with only parameters that are provided
235 | data = {}
236 |
237 | if params.label is not None:
238 | data["question_text"] = params.label
239 | if params.mandatory is not None:
240 | data["mandatory"] = str(params.mandatory).lower() # ServiceNow expects "true"/"false" strings
241 | if params.help_text is not None:
242 | data["help_text"] = params.help_text
243 | if params.default_value is not None:
244 | data["default_value"] = params.default_value
245 | if params.description is not None:
246 | data["description"] = params.description
247 | if params.order is not None:
248 | data["order"] = params.order
249 | if params.reference_qualifier is not None:
250 | data["reference_qual"] = params.reference_qualifier
251 | if params.max_length is not None:
252 | data["max_length"] = params.max_length
253 | if params.min is not None:
254 | data["min"] = params.min
255 | if params.max is not None:
256 | data["max"] = params.max
257 |
258 | # If no fields to update, return early
259 | if not data:
260 | return CatalogItemVariableResponse(
261 | success=False,
262 | message="No update parameters provided",
263 | )
264 |
265 | # Make request
266 | try:
267 | response = requests.patch(
268 | api_url,
269 | json=data,
270 | headers=auth_manager.get_headers(),
271 | timeout=config.timeout,
272 | )
273 | response.raise_for_status()
274 |
275 | result = response.json().get("result", {})
276 |
277 | return CatalogItemVariableResponse(
278 | success=True,
279 | message="Catalog item variable updated successfully",
280 | variable_id=params.variable_id,
281 | details=result,
282 | )
283 |
284 | except requests.RequestException as e:
285 | logger.error(f"Failed to update catalog item variable: {e}")
286 | return CatalogItemVariableResponse(
287 | success=False,
288 | message=f"Failed to update catalog item variable: {str(e)}",
289 | )
```
--------------------------------------------------------------------------------
/tests/test_change_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the change management tools.
3 | """
4 |
5 | import unittest
6 | from unittest.mock import MagicMock, patch
7 |
8 | import requests
9 |
10 | from servicenow_mcp.auth.auth_manager import AuthManager
11 | from servicenow_mcp.tools.change_tools import (
12 | create_change_request,
13 | list_change_requests,
14 | )
15 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
16 |
17 |
18 | class TestChangeTools(unittest.TestCase):
19 | """Tests for the change management tools."""
20 |
21 | def setUp(self):
22 | """Set up test fixtures."""
23 | self.auth_config = AuthConfig(
24 | type=AuthType.BASIC,
25 | basic=BasicAuthConfig(username="test_user", password="test_password"),
26 | )
27 | self.server_config = ServerConfig(
28 | instance_url="https://test.service-now.com",
29 | auth=self.auth_config,
30 | )
31 | self.auth_manager = AuthManager(self.auth_config)
32 |
33 | @patch("servicenow_mcp.tools.change_tools.requests.get")
34 | def test_list_change_requests_success(self, mock_get):
35 | """Test listing change requests successfully."""
36 | # Mock the response
37 | mock_response = MagicMock()
38 | mock_response.json.return_value = {
39 | "result": [
40 | {
41 | "sys_id": "change123",
42 | "number": "CHG0010001",
43 | "short_description": "Test Change",
44 | "type": "normal",
45 | "state": "open",
46 | },
47 | {
48 | "sys_id": "change456",
49 | "number": "CHG0010002",
50 | "short_description": "Another Test Change",
51 | "type": "emergency",
52 | "state": "in progress",
53 | },
54 | ]
55 | }
56 | mock_response.raise_for_status = MagicMock()
57 | mock_get.return_value = mock_response
58 |
59 | # Call the function
60 | params = {
61 | "limit": 10,
62 | "timeframe": "upcoming",
63 | }
64 | result = list_change_requests(self.auth_manager, self.server_config, params)
65 |
66 | # Verify the result
67 | self.assertTrue(result["success"])
68 | self.assertEqual(len(result["change_requests"]), 2)
69 | self.assertEqual(result["count"], 2)
70 | self.assertEqual(result["total"], 2)
71 | self.assertEqual(result["change_requests"][0]["sys_id"], "change123")
72 | self.assertEqual(result["change_requests"][1]["sys_id"], "change456")
73 |
74 | @patch("servicenow_mcp.tools.change_tools.requests.get")
75 | def test_list_change_requests_empty_result(self, mock_get):
76 | """Test listing change requests with empty result."""
77 | # Mock the response
78 | mock_response = MagicMock()
79 | mock_response.json.return_value = {"result": []}
80 | mock_response.raise_for_status = MagicMock()
81 | mock_get.return_value = mock_response
82 |
83 | # Call the function
84 | params = {
85 | "limit": 10,
86 | "timeframe": "upcoming",
87 | }
88 | result = list_change_requests(self.auth_manager, self.server_config, params)
89 |
90 | # Verify the result
91 | self.assertTrue(result["success"])
92 | self.assertEqual(len(result["change_requests"]), 0)
93 | self.assertEqual(result["count"], 0)
94 | self.assertEqual(result["total"], 0)
95 |
96 | @patch("servicenow_mcp.tools.change_tools.requests.get")
97 | def test_list_change_requests_missing_result(self, mock_get):
98 | """Test listing change requests with missing result key."""
99 | # Mock the response
100 | mock_response = MagicMock()
101 | mock_response.json.return_value = {} # No "result" key
102 | mock_response.raise_for_status = MagicMock()
103 | mock_get.return_value = mock_response
104 |
105 | # Call the function
106 | params = {
107 | "limit": 10,
108 | "timeframe": "upcoming",
109 | }
110 | result = list_change_requests(self.auth_manager, self.server_config, params)
111 |
112 | # Verify the result
113 | self.assertTrue(result["success"])
114 | self.assertEqual(len(result["change_requests"]), 0)
115 | self.assertEqual(result["count"], 0)
116 | self.assertEqual(result["total"], 0)
117 |
118 | @patch("servicenow_mcp.tools.change_tools.requests.get")
119 | def test_list_change_requests_error(self, mock_get):
120 | """Test listing change requests with error."""
121 | # Mock the response
122 | mock_get.side_effect = requests.exceptions.RequestException("Test error")
123 |
124 | # Call the function
125 | params = {
126 | "limit": 10,
127 | "timeframe": "upcoming",
128 | }
129 | result = list_change_requests(self.auth_manager, self.server_config, params)
130 |
131 | # Verify the result
132 | self.assertFalse(result["success"])
133 | self.assertIn("Error listing change requests", result["message"])
134 |
135 | @patch("servicenow_mcp.tools.change_tools.requests.get")
136 | def test_list_change_requests_with_filters(self, mock_get):
137 | """Test listing change requests with filters."""
138 | # Mock the response
139 | mock_response = MagicMock()
140 | mock_response.json.return_value = {
141 | "result": [
142 | {
143 | "sys_id": "change123",
144 | "number": "CHG0010001",
145 | "short_description": "Test Change",
146 | "type": "normal",
147 | "state": "open",
148 | }
149 | ]
150 | }
151 | mock_response.raise_for_status = MagicMock()
152 | mock_get.return_value = mock_response
153 |
154 | # Call the function with filters
155 | params = {
156 | "limit": 10,
157 | "state": "open",
158 | "type": "normal",
159 | "category": "Hardware",
160 | "assignment_group": "IT Support",
161 | "timeframe": "upcoming",
162 | "query": "short_description=Test",
163 | }
164 | result = list_change_requests(self.auth_manager, self.server_config, params)
165 |
166 | # Verify the result
167 | self.assertTrue(result["success"])
168 | self.assertEqual(len(result["change_requests"]), 1)
169 |
170 | # Verify that the correct query parameters were passed to the request
171 | args, kwargs = mock_get.call_args
172 | self.assertIn("params", kwargs)
173 | self.assertIn("sysparm_query", kwargs["params"])
174 | query = kwargs["params"]["sysparm_query"]
175 |
176 | # Check that all filters are in the query
177 | self.assertIn("state=open", query)
178 | self.assertIn("type=normal", query)
179 | self.assertIn("category=Hardware", query)
180 | self.assertIn("assignment_group=IT Support", query)
181 | self.assertIn("short_description=Test", query)
182 | # The timeframe filter adds a date comparison, which is harder to test exactly
183 |
184 | @patch("servicenow_mcp.tools.change_tools.requests.post")
185 | def test_create_change_request_with_swapped_parameters(self, mock_post):
186 | """Test creating a change request with swapped parameters (server_config used as auth_manager)."""
187 | # Mock the response
188 | mock_response = MagicMock()
189 | mock_response.json.return_value = {
190 | "result": {
191 | "sys_id": "change123",
192 | "number": "CHG0010001",
193 | "short_description": "Test Change",
194 | "type": "normal",
195 | }
196 | }
197 | mock_response.raise_for_status = MagicMock()
198 | mock_post.return_value = mock_response
199 |
200 | # Create a server_config with a get_headers method to simulate what might happen in Claude Desktop
201 | server_config_with_headers = MagicMock()
202 | server_config_with_headers.instance_url = "https://test.service-now.com"
203 | server_config_with_headers.get_headers.return_value = {"Authorization": "Basic dGVzdF91c2VyOnRlc3RfcGFzc3dvcmQ="}
204 |
205 | # Call the function with swapped parameters (server_config as auth_manager)
206 | params = {
207 | "short_description": "Test Change",
208 | "type": "normal",
209 | "risk": "low",
210 | "impact": "medium",
211 | }
212 | result = create_change_request(server_config_with_headers, self.auth_manager, params)
213 |
214 | # Verify the result
215 | self.assertTrue(result["success"])
216 | self.assertEqual(result["change_request"]["sys_id"], "change123")
217 | self.assertEqual(result["change_request"]["number"], "CHG0010001")
218 |
219 | @patch("servicenow_mcp.tools.change_tools.requests.post")
220 | def test_create_change_request_with_serverconfig_no_get_headers(self, mock_post):
221 | """Test creating a change request with ServerConfig object that doesn't have get_headers method."""
222 | # This test simulates the exact error we're seeing in Claude Desktop
223 |
224 | # Create params for the change request
225 | params = {
226 | "short_description": "Test Change",
227 | "type": "normal",
228 | "risk": "low",
229 | "impact": "medium",
230 | }
231 |
232 | # Create a real ServerConfig object (which doesn't have get_headers method)
233 | # and a mock AuthManager object (which doesn't have instance_url)
234 | real_server_config = ServerConfig(
235 | instance_url="https://test.service-now.com",
236 | auth=self.auth_config,
237 | )
238 |
239 | mock_auth_manager = MagicMock()
240 | # Explicitly remove get_headers method to simulate the error
241 | if hasattr(mock_auth_manager, 'get_headers'):
242 | delattr(mock_auth_manager, 'get_headers')
243 |
244 | # Call the function with parameters that will cause the error
245 | result = create_change_request(real_server_config, mock_auth_manager, params)
246 |
247 | # The function should detect the issue and return an error message
248 | self.assertFalse(result["success"])
249 | self.assertIn("Cannot find get_headers method", result["message"])
250 |
251 | # Verify that the post method was never called
252 | mock_post.assert_not_called()
253 |
254 | @patch("servicenow_mcp.tools.change_tools.requests.post")
255 | def test_create_change_request_with_swapped_parameters_real(self, mock_post):
256 | """Test creating a change request with swapped parameters (auth_manager and server_config)."""
257 | # Mock the response
258 | mock_response = MagicMock()
259 | mock_response.json.return_value = {
260 | "result": {
261 | "sys_id": "change123",
262 | "number": "CHG0010001",
263 | "short_description": "Test Change",
264 | "type": "normal",
265 | }
266 | }
267 | mock_response.raise_for_status = MagicMock()
268 | mock_post.return_value = mock_response
269 |
270 | # Create params for the change request
271 | params = {
272 | "short_description": "Test Change",
273 | "type": "normal",
274 | "risk": "low",
275 | "impact": "medium",
276 | }
277 |
278 | # Call the function with swapped parameters (server_config as first parameter, auth_manager as second)
279 | result = create_change_request(self.server_config, self.auth_manager, params)
280 |
281 | # The function should still work correctly
282 | self.assertTrue(result["success"])
283 | self.assertEqual(result["change_request"]["sys_id"], "change123")
284 | self.assertEqual(result["change_request"]["number"], "CHG0010001")
285 |
286 |
287 | if __name__ == "__main__":
288 | unittest.main()
```
--------------------------------------------------------------------------------
/tests/test_catalog_resources.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the ServiceNow MCP catalog resources.
3 | """
4 |
5 | import unittest
6 | from unittest.mock import MagicMock, patch
7 |
8 | from servicenow_mcp.auth.auth_manager import AuthManager
9 | from servicenow_mcp.resources.catalog import (
10 | CatalogCategoryListParams,
11 | CatalogItemVariableModel,
12 | CatalogListParams,
13 | CatalogResource,
14 | )
15 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
16 |
17 |
18 | class TestCatalogResource(unittest.TestCase):
19 | """Test cases for the catalog resource."""
20 |
21 | def setUp(self):
22 | """Set up test fixtures."""
23 | # Create a mock server config
24 | self.config = ServerConfig(
25 | instance_url="https://example.service-now.com",
26 | auth=AuthConfig(
27 | type=AuthType.BASIC,
28 | basic=BasicAuthConfig(username="admin", password="password"),
29 | ),
30 | )
31 |
32 | # Create a mock auth manager
33 | self.auth_manager = MagicMock(spec=AuthManager)
34 | self.auth_manager.get_headers.return_value = {"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}
35 |
36 | # Create the resource
37 | self.resource = CatalogResource(self.config, self.auth_manager)
38 |
39 | @patch("servicenow_mcp.resources.catalog.requests.get")
40 | async def test_list_catalog_items(self, mock_get):
41 | """Test listing catalog items."""
42 | # Mock the response from ServiceNow
43 | mock_response = MagicMock()
44 | mock_response.json.return_value = {
45 | "result": [
46 | {
47 | "sys_id": "item1",
48 | "name": "Laptop",
49 | "short_description": "Request a new laptop",
50 | "category": "Hardware",
51 | "price": "1000",
52 | "picture": "laptop.jpg",
53 | "active": "true",
54 | "order": "100",
55 | }
56 | ]
57 | }
58 | mock_response.raise_for_status = MagicMock()
59 | mock_get.return_value = mock_response
60 |
61 | # Call the method
62 | params = CatalogListParams(
63 | limit=10,
64 | offset=0,
65 | category="Hardware",
66 | query="laptop",
67 | )
68 | result = await self.resource.list_catalog_items(params)
69 |
70 | # Check the result
71 | self.assertEqual(len(result), 1)
72 | self.assertEqual(result[0].name, "Laptop")
73 | self.assertEqual(result[0].category, "Hardware")
74 | self.assertTrue(result[0].active)
75 |
76 | # Check that the correct URL and parameters were used
77 | mock_get.assert_called_once()
78 | args, kwargs = mock_get.call_args
79 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item")
80 | self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
81 | self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
82 | self.assertIn("sysparm_query", kwargs["params"])
83 | self.assertIn("active=true", kwargs["params"]["sysparm_query"])
84 | self.assertIn("category=Hardware", kwargs["params"]["sysparm_query"])
85 | self.assertIn("short_descriptionLIKElaptop^ORnameLIKElaptop", kwargs["params"]["sysparm_query"])
86 |
87 | @patch("servicenow_mcp.resources.catalog.requests.get")
88 | async def test_list_catalog_items_error(self, mock_get):
89 | """Test listing catalog items with an error."""
90 | # Mock the response from ServiceNow
91 | mock_get.side_effect = Exception("Error")
92 |
93 | # Call the method
94 | params = CatalogListParams(
95 | limit=10,
96 | offset=0,
97 | )
98 | result = await self.resource.list_catalog_items(params)
99 |
100 | # Check the result
101 | self.assertEqual(len(result), 0)
102 |
103 | @patch("servicenow_mcp.resources.catalog.CatalogResource.get_catalog_item_variables")
104 | @patch("servicenow_mcp.resources.catalog.requests.get")
105 | async def test_get_catalog_item(self, mock_get, mock_get_variables):
106 | """Test getting a specific catalog item."""
107 | # Mock the response from ServiceNow
108 | mock_response = MagicMock()
109 | mock_response.json.return_value = {
110 | "result": {
111 | "sys_id": "item1",
112 | "name": "Laptop",
113 | "short_description": "Request a new laptop",
114 | "description": "Request a new laptop for work",
115 | "category": "Hardware",
116 | "price": "1000",
117 | "picture": "laptop.jpg",
118 | "active": "true",
119 | "order": "100",
120 | "delivery_time": "3 days",
121 | "availability": "In Stock",
122 | }
123 | }
124 | mock_response.raise_for_status = MagicMock()
125 | mock_get.return_value = mock_response
126 |
127 | # Mock the variables
128 | mock_get_variables.return_value = [
129 | CatalogItemVariableModel(
130 | sys_id="var1",
131 | name="model",
132 | label="Laptop Model",
133 | type="string",
134 | mandatory=True,
135 | default_value="MacBook Pro",
136 | help_text="Select the laptop model",
137 | order=100,
138 | )
139 | ]
140 |
141 | # Call the method
142 | result = await self.resource.get_catalog_item("item1")
143 |
144 | # Check the result
145 | self.assertEqual(result["sys_id"], "item1")
146 | self.assertEqual(result["name"], "Laptop")
147 | self.assertEqual(result["category"], "Hardware")
148 | self.assertTrue(result["active"])
149 | self.assertEqual(len(result["variables"]), 1)
150 | self.assertEqual(result["variables"][0].name, "model")
151 |
152 | # Check that the correct URL and parameters were used
153 | mock_get.assert_called_once()
154 | args, kwargs = mock_get.call_args
155 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item/item1")
156 |
157 | @patch("servicenow_mcp.resources.catalog.requests.get")
158 | async def test_get_catalog_item_not_found(self, mock_get):
159 | """Test getting a catalog item that doesn't exist."""
160 | # Mock the response from ServiceNow
161 | mock_response = MagicMock()
162 | mock_response.json.return_value = {"result": {}}
163 | mock_response.raise_for_status = MagicMock()
164 | mock_get.return_value = mock_response
165 |
166 | # Call the method
167 | result = await self.resource.get_catalog_item("nonexistent")
168 |
169 | # Check the result
170 | self.assertIn("error", result)
171 | self.assertIn("not found", result["error"])
172 |
173 | @patch("servicenow_mcp.resources.catalog.requests.get")
174 | async def test_get_catalog_item_error(self, mock_get):
175 | """Test getting a catalog item with an error."""
176 | # Mock the response from ServiceNow
177 | mock_get.side_effect = Exception("Error")
178 |
179 | # Call the method
180 | result = await self.resource.get_catalog_item("item1")
181 |
182 | # Check the result
183 | self.assertIn("error", result)
184 | self.assertIn("Error", result["error"])
185 |
186 | @patch("servicenow_mcp.resources.catalog.requests.get")
187 | async def test_get_catalog_item_variables(self, mock_get):
188 | """Test getting variables for a catalog item."""
189 | # Mock the response from ServiceNow
190 | mock_response = MagicMock()
191 | mock_response.json.return_value = {
192 | "result": [
193 | {
194 | "sys_id": "var1",
195 | "name": "model",
196 | "question_text": "Laptop Model",
197 | "type": "string",
198 | "mandatory": "true",
199 | "default_value": "MacBook Pro",
200 | "help_text": "Select the laptop model",
201 | "order": "100",
202 | }
203 | ]
204 | }
205 | mock_response.raise_for_status = MagicMock()
206 | mock_get.return_value = mock_response
207 |
208 | # Call the method
209 | result = await self.resource.get_catalog_item_variables("item1")
210 |
211 | # Check the result
212 | self.assertEqual(len(result), 1)
213 | self.assertEqual(result[0].name, "model")
214 | self.assertEqual(result[0].label, "Laptop Model")
215 | self.assertEqual(result[0].type, "string")
216 | self.assertTrue(result[0].mandatory)
217 |
218 | # Check that the correct URL and parameters were used
219 | mock_get.assert_called_once()
220 | args, kwargs = mock_get.call_args
221 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/item_option_new")
222 | self.assertEqual(kwargs["params"]["sysparm_query"], "cat_item=item1^ORDERBYorder")
223 |
224 | @patch("servicenow_mcp.resources.catalog.requests.get")
225 | async def test_get_catalog_item_variables_error(self, mock_get):
226 | """Test getting variables for a catalog item with an error."""
227 | # Mock the response from ServiceNow
228 | mock_get.side_effect = Exception("Error")
229 |
230 | # Call the method
231 | result = await self.resource.get_catalog_item_variables("item1")
232 |
233 | # Check the result
234 | self.assertEqual(len(result), 0)
235 |
236 | @patch("servicenow_mcp.resources.catalog.requests.get")
237 | async def test_list_catalog_categories(self, mock_get):
238 | """Test listing catalog categories."""
239 | # Mock the response from ServiceNow
240 | mock_response = MagicMock()
241 | mock_response.json.return_value = {
242 | "result": [
243 | {
244 | "sys_id": "cat1",
245 | "title": "Hardware",
246 | "description": "Hardware requests",
247 | "parent": "",
248 | "icon": "hardware.png",
249 | "active": "true",
250 | "order": "100",
251 | }
252 | ]
253 | }
254 | mock_response.raise_for_status = MagicMock()
255 | mock_get.return_value = mock_response
256 |
257 | # Call the method
258 | params = CatalogCategoryListParams(
259 | limit=10,
260 | offset=0,
261 | query="hardware",
262 | )
263 | result = await self.resource.list_catalog_categories(params)
264 |
265 | # Check the result
266 | self.assertEqual(len(result), 1)
267 | self.assertEqual(result[0].title, "Hardware")
268 | self.assertEqual(result[0].description, "Hardware requests")
269 | self.assertTrue(result[0].active)
270 |
271 | # Check that the correct URL and parameters were used
272 | mock_get.assert_called_once()
273 | args, kwargs = mock_get.call_args
274 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_category")
275 | self.assertEqual(kwargs["params"]["sysparm_limit"], 10)
276 | self.assertEqual(kwargs["params"]["sysparm_offset"], 0)
277 | self.assertIn("sysparm_query", kwargs["params"])
278 | self.assertIn("active=true", kwargs["params"]["sysparm_query"])
279 | self.assertIn("titleLIKEhardware^ORdescriptionLIKEhardware", kwargs["params"]["sysparm_query"])
280 |
281 | @patch("servicenow_mcp.resources.catalog.requests.get")
282 | async def test_list_catalog_categories_error(self, mock_get):
283 | """Test listing catalog categories with an error."""
284 | # Mock the response from ServiceNow
285 | mock_get.side_effect = Exception("Error")
286 |
287 | # Call the method
288 | params = CatalogCategoryListParams(
289 | limit=10,
290 | offset=0,
291 | )
292 | result = await self.resource.list_catalog_categories(params)
293 |
294 | # Check the result
295 | self.assertEqual(len(result), 0)
296 |
297 | @patch("servicenow_mcp.resources.catalog.CatalogResource.get_catalog_item")
298 | async def test_read(self, mock_get_catalog_item):
299 | """Test reading a catalog item."""
300 | # Mock the get_catalog_item method
301 | mock_get_catalog_item.return_value = {
302 | "sys_id": "item1",
303 | "name": "Laptop",
304 | }
305 |
306 | # Call the method
307 | result = await self.resource.read({"item_id": "item1"})
308 |
309 | # Check the result
310 | self.assertEqual(result["sys_id"], "item1")
311 | self.assertEqual(result["name"], "Laptop")
312 |
313 | # Check that the correct method was called
314 | mock_get_catalog_item.assert_called_once_with("item1")
315 |
316 | async def test_read_missing_param(self):
317 | """Test reading a catalog item with missing parameter."""
318 | # Call the method
319 | result = await self.resource.read({})
320 |
321 | # Check the result
322 | self.assertIn("error", result)
323 | self.assertIn("Missing item_id parameter", result["error"])
324 |
325 |
326 | if __name__ == "__main__":
327 | unittest.main()
```