This is page 4 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
--------------------------------------------------------------------------------
/tests/test_workflow_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the workflow management tools.
3 | """
4 |
5 | import json
6 | import unittest
7 | from datetime import datetime, timedelta
8 | from unittest.mock import MagicMock, patch
9 |
10 | import requests
11 |
12 | from servicenow_mcp.auth.auth_manager import AuthManager
13 | from servicenow_mcp.tools.workflow_tools import (
14 | list_workflows,
15 | get_workflow_details,
16 | list_workflow_versions,
17 | get_workflow_activities,
18 | create_workflow,
19 | update_workflow,
20 | activate_workflow,
21 | deactivate_workflow,
22 | add_workflow_activity,
23 | update_workflow_activity,
24 | delete_workflow_activity,
25 | reorder_workflow_activities,
26 | )
27 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
28 |
29 |
30 | class TestWorkflowTools(unittest.TestCase):
31 | """Tests for the workflow management tools."""
32 |
33 | def setUp(self):
34 | """Set up test fixtures."""
35 | self.auth_config = AuthConfig(
36 | type=AuthType.BASIC,
37 | basic=BasicAuthConfig(username="test_user", password="test_password"),
38 | )
39 | self.server_config = ServerConfig(
40 | instance_url="https://test.service-now.com",
41 | auth=self.auth_config,
42 | )
43 | self.auth_manager = AuthManager(self.auth_config)
44 |
45 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
46 | def test_list_workflows_success(self, mock_get):
47 | """Test listing workflows successfully."""
48 | # Mock the response
49 | mock_response = MagicMock()
50 | mock_response.json.return_value = {
51 | "result": [
52 | {
53 | "sys_id": "workflow123",
54 | "name": "Incident Approval",
55 | "description": "Workflow for incident approval",
56 | "active": "true",
57 | "table": "incident",
58 | },
59 | {
60 | "sys_id": "workflow456",
61 | "name": "Change Request",
62 | "description": "Workflow for change requests",
63 | "active": "true",
64 | "table": "change_request",
65 | },
66 | ]
67 | }
68 | mock_response.headers = {"X-Total-Count": "2"}
69 | mock_response.raise_for_status = MagicMock()
70 | mock_get.return_value = mock_response
71 |
72 | # Call the function
73 | params = {
74 | "limit": 10,
75 | "active": True,
76 | }
77 | result = list_workflows(self.auth_manager, self.server_config, params)
78 |
79 | # Verify the result
80 | self.assertEqual(len(result["workflows"]), 2)
81 | self.assertEqual(result["count"], 2)
82 | self.assertEqual(result["total"], 2)
83 | self.assertEqual(result["workflows"][0]["sys_id"], "workflow123")
84 | self.assertEqual(result["workflows"][1]["sys_id"], "workflow456")
85 |
86 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
87 | def test_list_workflows_empty_result(self, mock_get):
88 | """Test listing workflows with empty result."""
89 | # Mock the response
90 | mock_response = MagicMock()
91 | mock_response.json.return_value = {"result": []}
92 | mock_response.headers = {"X-Total-Count": "0"}
93 | mock_response.raise_for_status = MagicMock()
94 | mock_get.return_value = mock_response
95 |
96 | # Call the function
97 | params = {
98 | "limit": 10,
99 | "active": True,
100 | }
101 | result = list_workflows(self.auth_manager, self.server_config, params)
102 |
103 | # Verify the result
104 | self.assertEqual(len(result["workflows"]), 0)
105 | self.assertEqual(result["count"], 0)
106 | self.assertEqual(result["total"], 0)
107 |
108 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
109 | def test_list_workflows_error(self, mock_get):
110 | """Test listing workflows with error."""
111 | # Mock the response
112 | mock_get.side_effect = requests.RequestException("API Error")
113 |
114 | # Call the function
115 | params = {
116 | "limit": 10,
117 | "active": True,
118 | }
119 | result = list_workflows(self.auth_manager, self.server_config, params)
120 |
121 | # Verify the result
122 | self.assertIn("error", result)
123 | self.assertEqual(result["error"], "API Error")
124 |
125 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
126 | def test_get_workflow_details_success(self, mock_get):
127 | """Test getting workflow details successfully."""
128 | # Mock the response
129 | mock_response = MagicMock()
130 | mock_response.json.return_value = {
131 | "result": {
132 | "sys_id": "workflow123",
133 | "name": "Incident Approval",
134 | "description": "Workflow for incident approval",
135 | "active": "true",
136 | "table": "incident",
137 | }
138 | }
139 | mock_response.raise_for_status = MagicMock()
140 | mock_get.return_value = mock_response
141 |
142 | # Call the function
143 | params = {
144 | "workflow_id": "workflow123",
145 | }
146 | result = get_workflow_details(self.auth_manager, self.server_config, params)
147 |
148 | # Verify the result
149 | self.assertEqual(result["workflow"]["sys_id"], "workflow123")
150 | self.assertEqual(result["workflow"]["name"], "Incident Approval")
151 |
152 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
153 | def test_get_workflow_details_error(self, mock_get):
154 | """Test getting workflow details with error."""
155 | # Mock the response
156 | mock_get.side_effect = requests.RequestException("API Error")
157 |
158 | # Call the function
159 | params = {
160 | "workflow_id": "workflow123",
161 | }
162 | result = get_workflow_details(self.auth_manager, self.server_config, params)
163 |
164 | # Verify the result
165 | self.assertIn("error", result)
166 | self.assertEqual(result["error"], "API Error")
167 |
168 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
169 | def test_list_workflow_versions_success(self, mock_get):
170 | """Test listing workflow versions successfully."""
171 | # Mock the response
172 | mock_response = MagicMock()
173 | mock_response.json.return_value = {
174 | "result": [
175 | {
176 | "sys_id": "version123",
177 | "workflow": "workflow123",
178 | "name": "Version 1",
179 | "version": "1",
180 | "published": "true",
181 | },
182 | {
183 | "sys_id": "version456",
184 | "workflow": "workflow123",
185 | "name": "Version 2",
186 | "version": "2",
187 | "published": "true",
188 | },
189 | ]
190 | }
191 | mock_response.headers = {"X-Total-Count": "2"}
192 | mock_response.raise_for_status = MagicMock()
193 | mock_get.return_value = mock_response
194 |
195 | # Call the function
196 | params = {
197 | "workflow_id": "workflow123",
198 | "limit": 10,
199 | }
200 | result = list_workflow_versions(self.auth_manager, self.server_config, params)
201 |
202 | # Verify the result
203 | self.assertEqual(len(result["versions"]), 2)
204 | self.assertEqual(result["count"], 2)
205 | self.assertEqual(result["total"], 2)
206 | self.assertEqual(result["versions"][0]["sys_id"], "version123")
207 | self.assertEqual(result["versions"][1]["sys_id"], "version456")
208 |
209 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
210 | def test_get_workflow_activities_success(self, mock_get):
211 | """Test getting workflow activities successfully."""
212 | # Mock the responses for version query and activities query
213 | version_response = MagicMock()
214 | version_response.json.return_value = {
215 | "result": [
216 | {
217 | "sys_id": "version123",
218 | "workflow": "workflow123",
219 | "name": "Version 1",
220 | "version": "1",
221 | "published": "true",
222 | }
223 | ]
224 | }
225 | version_response.raise_for_status = MagicMock()
226 |
227 | activities_response = MagicMock()
228 | activities_response.json.return_value = {
229 | "result": [
230 | {
231 | "sys_id": "activity123",
232 | "workflow_version": "version123",
233 | "name": "Approval",
234 | "order": "100",
235 | "activity_definition": "approval",
236 | },
237 | {
238 | "sys_id": "activity456",
239 | "workflow_version": "version123",
240 | "name": "Notification",
241 | "order": "200",
242 | "activity_definition": "notification",
243 | },
244 | ]
245 | }
246 | activities_response.raise_for_status = MagicMock()
247 |
248 | # Configure the mock to return different responses for different URLs
249 | def side_effect(*args, **kwargs):
250 | url = args[0] if args else kwargs.get('url', '')
251 | if 'wf_workflow_version' in url:
252 | return version_response
253 | elif 'wf_activity' in url:
254 | return activities_response
255 | return MagicMock()
256 |
257 | mock_get.side_effect = side_effect
258 |
259 | # Call the function
260 | params = {
261 | "workflow_id": "workflow123",
262 | }
263 | result = get_workflow_activities(self.auth_manager, self.server_config, params)
264 |
265 | # Verify the result
266 | self.assertEqual(len(result["activities"]), 2)
267 | self.assertEqual(result["count"], 2)
268 | self.assertEqual(result["workflow_id"], "workflow123")
269 | self.assertEqual(result["version_id"], "version123")
270 | self.assertEqual(result["activities"][0]["sys_id"], "activity123")
271 | self.assertEqual(result["activities"][1]["sys_id"], "activity456")
272 |
273 | @patch("servicenow_mcp.tools.workflow_tools.requests.post")
274 | def test_create_workflow_success(self, mock_post):
275 | """Test creating a workflow successfully."""
276 | # Mock the response
277 | mock_response = MagicMock()
278 | mock_response.json.return_value = {
279 | "result": {
280 | "sys_id": "workflow789",
281 | "name": "New Workflow",
282 | "description": "A new workflow",
283 | "active": "true",
284 | "table": "incident",
285 | }
286 | }
287 | mock_response.raise_for_status = MagicMock()
288 | mock_post.return_value = mock_response
289 |
290 | # Call the function
291 | params = {
292 | "name": "New Workflow",
293 | "description": "A new workflow",
294 | "table": "incident",
295 | "active": True,
296 | }
297 | result = create_workflow(self.auth_manager, self.server_config, params)
298 |
299 | # Verify the result
300 | self.assertEqual(result["workflow"]["sys_id"], "workflow789")
301 | self.assertEqual(result["workflow"]["name"], "New Workflow")
302 | self.assertEqual(result["message"], "Workflow created successfully")
303 |
304 | @patch("servicenow_mcp.tools.workflow_tools.requests.patch")
305 | def test_update_workflow_success(self, mock_patch):
306 | """Test updating a workflow successfully."""
307 | # Mock the response
308 | mock_response = MagicMock()
309 | mock_response.json.return_value = {
310 | "result": {
311 | "sys_id": "workflow123",
312 | "name": "Updated Workflow",
313 | "description": "Updated description",
314 | "active": "true",
315 | "table": "incident",
316 | }
317 | }
318 | mock_response.raise_for_status = MagicMock()
319 | mock_patch.return_value = mock_response
320 |
321 | # Call the function
322 | params = {
323 | "workflow_id": "workflow123",
324 | "name": "Updated Workflow",
325 | "description": "Updated description",
326 | }
327 | result = update_workflow(self.auth_manager, self.server_config, params)
328 |
329 | # Verify the result
330 | self.assertEqual(result["workflow"]["sys_id"], "workflow123")
331 | self.assertEqual(result["workflow"]["name"], "Updated Workflow")
332 | self.assertEqual(result["message"], "Workflow updated successfully")
333 |
334 | @patch("servicenow_mcp.tools.workflow_tools.requests.patch")
335 | def test_activate_workflow_success(self, mock_patch):
336 | """Test activating a workflow successfully."""
337 | # Mock the response
338 | mock_response = MagicMock()
339 | mock_response.json.return_value = {
340 | "result": {
341 | "sys_id": "workflow123",
342 | "name": "Incident Approval",
343 | "active": "true",
344 | }
345 | }
346 | mock_response.raise_for_status = MagicMock()
347 | mock_patch.return_value = mock_response
348 |
349 | # Call the function
350 | params = {
351 | "workflow_id": "workflow123",
352 | }
353 | result = activate_workflow(self.auth_manager, self.server_config, params)
354 |
355 | # Verify the result
356 | self.assertEqual(result["workflow"]["sys_id"], "workflow123")
357 | self.assertEqual(result["workflow"]["active"], "true")
358 | self.assertEqual(result["message"], "Workflow activated successfully")
359 |
360 | @patch("servicenow_mcp.tools.workflow_tools.requests.patch")
361 | def test_deactivate_workflow_success(self, mock_patch):
362 | """Test deactivating a workflow successfully."""
363 | # Mock the response
364 | mock_response = MagicMock()
365 | mock_response.json.return_value = {
366 | "result": {
367 | "sys_id": "workflow123",
368 | "name": "Incident Approval",
369 | "active": "false",
370 | }
371 | }
372 | mock_response.raise_for_status = MagicMock()
373 | mock_patch.return_value = mock_response
374 |
375 | # Call the function
376 | params = {
377 | "workflow_id": "workflow123",
378 | }
379 | result = deactivate_workflow(self.auth_manager, self.server_config, params)
380 |
381 | # Verify the result
382 | self.assertEqual(result["workflow"]["sys_id"], "workflow123")
383 | self.assertEqual(result["workflow"]["active"], "false")
384 | self.assertEqual(result["message"], "Workflow deactivated successfully")
385 |
386 | @patch("servicenow_mcp.tools.workflow_tools.requests.get")
387 | @patch("servicenow_mcp.tools.workflow_tools.requests.post")
388 | def test_add_workflow_activity_success(self, mock_post, mock_get):
389 | """Test adding a workflow activity successfully."""
390 | # Mock the responses for version query and activity creation
391 | version_response = MagicMock()
392 | version_response.json.return_value = {
393 | "result": [
394 | {
395 | "sys_id": "version123",
396 | "workflow": "workflow123",
397 | "name": "Version 1",
398 | "version": "1",
399 | "published": "false",
400 | }
401 | ]
402 | }
403 | version_response.raise_for_status = MagicMock()
404 |
405 | order_response = MagicMock()
406 | order_response.json.return_value = {
407 | "result": [
408 | {
409 | "sys_id": "activity123",
410 | "order": "100",
411 | }
412 | ]
413 | }
414 | order_response.raise_for_status = MagicMock()
415 |
416 | activity_response = MagicMock()
417 | activity_response.json.return_value = {
418 | "result": {
419 | "sys_id": "activity789",
420 | "workflow_version": "version123",
421 | "name": "New Activity",
422 | "order": "200",
423 | "activity_definition": "approval",
424 | }
425 | }
426 | activity_response.raise_for_status = MagicMock()
427 |
428 | # Configure the mocks
429 | def get_side_effect(*args, **kwargs):
430 | url = args[0] if args else kwargs.get('url', '')
431 | if 'wf_workflow_version' in url:
432 | return version_response
433 | elif 'wf_activity' in url:
434 | return order_response
435 | return MagicMock()
436 |
437 | mock_get.side_effect = get_side_effect
438 | mock_post.return_value = activity_response
439 |
440 | # Call the function
441 | params = {
442 | "workflow_id": "workflow123",
443 | "name": "New Activity",
444 | "activity_type": "approval",
445 | "description": "A new approval activity",
446 | }
447 | result = add_workflow_activity(self.auth_manager, self.server_config, params)
448 |
449 | # Verify the result
450 | self.assertEqual(result["activity"]["sys_id"], "activity789")
451 | self.assertEqual(result["activity"]["name"], "New Activity")
452 | self.assertEqual(result["workflow_id"], "workflow123")
453 | self.assertEqual(result["version_id"], "version123")
454 | self.assertEqual(result["message"], "Activity added successfully")
455 |
456 | @patch("servicenow_mcp.tools.workflow_tools.requests.patch")
457 | def test_update_workflow_activity_success(self, mock_patch):
458 | """Test updating a workflow activity successfully."""
459 | # Mock the response
460 | mock_response = MagicMock()
461 | mock_response.json.return_value = {
462 | "result": {
463 | "sys_id": "activity123",
464 | "name": "Updated Activity",
465 | "description": "Updated description",
466 | }
467 | }
468 | mock_response.raise_for_status = MagicMock()
469 | mock_patch.return_value = mock_response
470 |
471 | # Call the function
472 | params = {
473 | "activity_id": "activity123",
474 | "name": "Updated Activity",
475 | "description": "Updated description",
476 | }
477 | result = update_workflow_activity(self.auth_manager, self.server_config, params)
478 |
479 | # Verify the result
480 | self.assertEqual(result["activity"]["sys_id"], "activity123")
481 | self.assertEqual(result["activity"]["name"], "Updated Activity")
482 | self.assertEqual(result["message"], "Activity updated successfully")
483 |
484 | @patch("servicenow_mcp.tools.workflow_tools.requests.delete")
485 | def test_delete_workflow_activity_success(self, mock_delete):
486 | """Test deleting a workflow activity successfully."""
487 | # Mock the response
488 | mock_response = MagicMock()
489 | mock_response.raise_for_status = MagicMock()
490 | mock_delete.return_value = mock_response
491 |
492 | # Call the function
493 | params = {
494 | "activity_id": "activity123",
495 | }
496 | result = delete_workflow_activity(self.auth_manager, self.server_config, params)
497 |
498 | # Verify the result
499 | self.assertEqual(result["message"], "Activity deleted successfully")
500 | self.assertEqual(result["activity_id"], "activity123")
501 |
502 | @patch("servicenow_mcp.tools.workflow_tools.requests.patch")
503 | def test_reorder_workflow_activities_success(self, mock_patch):
504 | """Test reordering workflow activities successfully."""
505 | # Mock the response
506 | mock_response = MagicMock()
507 | mock_response.json.return_value = {"result": {}}
508 | mock_response.raise_for_status = MagicMock()
509 | mock_patch.return_value = mock_response
510 |
511 | # Call the function
512 | params = {
513 | "workflow_id": "workflow123",
514 | "activity_ids": ["activity1", "activity2", "activity3"],
515 | }
516 | result = reorder_workflow_activities(self.auth_manager, self.server_config, params)
517 |
518 | # Verify the result
519 | self.assertEqual(result["message"], "Activities reordered")
520 | self.assertEqual(result["workflow_id"], "workflow123")
521 | self.assertEqual(len(result["results"]), 3)
522 | self.assertTrue(all(item["success"] for item in result["results"]))
523 | self.assertEqual(result["results"][0]["activity_id"], "activity1")
524 | self.assertEqual(result["results"][0]["new_order"], 100)
525 | self.assertEqual(result["results"][1]["activity_id"], "activity2")
526 | self.assertEqual(result["results"][1]["new_order"], 200)
527 | self.assertEqual(result["results"][2]["activity_id"], "activity3")
528 | self.assertEqual(result["results"][2]["new_order"], 300)
529 |
530 |
531 | if __name__ == "__main__":
532 | unittest.main()
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/catalog_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Service Catalog tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for querying and viewing the service catalog in ServiceNow.
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 ListCatalogItemsParams(BaseModel):
20 | """Parameters for listing service catalog items."""
21 |
22 | limit: int = Field(10, description="Maximum number of catalog items to return")
23 | offset: int = Field(0, description="Offset for pagination")
24 | category: Optional[str] = Field(None, description="Filter by category")
25 | query: Optional[str] = Field(None, description="Search query for catalog items")
26 | active: bool = Field(True, description="Whether to only return active catalog items")
27 |
28 |
29 | class GetCatalogItemParams(BaseModel):
30 | """Parameters for getting a specific service catalog item."""
31 |
32 | item_id: str = Field(..., description="Catalog item ID or sys_id")
33 |
34 |
35 | class ListCatalogCategoriesParams(BaseModel):
36 | """Parameters for listing service catalog categories."""
37 |
38 | limit: int = Field(10, description="Maximum number of categories to return")
39 | offset: int = Field(0, description="Offset for pagination")
40 | query: Optional[str] = Field(None, description="Search query for categories")
41 | active: bool = Field(True, description="Whether to only return active categories")
42 |
43 |
44 | class CatalogResponse(BaseModel):
45 | """Response from catalog operations."""
46 |
47 | success: bool = Field(..., description="Whether the operation was successful")
48 | message: str = Field(..., description="Message describing the result")
49 | data: Optional[Dict[str, Any]] = Field(None, description="Response data")
50 |
51 |
52 | class CreateCatalogCategoryParams(BaseModel):
53 | """Parameters for creating a new service catalog category."""
54 |
55 | title: str = Field(..., description="Title of the category")
56 | description: Optional[str] = Field(None, description="Description of the category")
57 | parent: Optional[str] = Field(None, description="Parent category sys_id")
58 | icon: Optional[str] = Field(None, description="Icon for the category")
59 | active: bool = Field(True, description="Whether the category is active")
60 | order: Optional[int] = Field(None, description="Order of the category")
61 |
62 |
63 | class UpdateCatalogCategoryParams(BaseModel):
64 | """Parameters for updating a service catalog category."""
65 |
66 | category_id: str = Field(..., description="Category ID or sys_id")
67 | title: Optional[str] = Field(None, description="Title of the category")
68 | description: Optional[str] = Field(None, description="Description of the category")
69 | parent: Optional[str] = Field(None, description="Parent category sys_id")
70 | icon: Optional[str] = Field(None, description="Icon for the category")
71 | active: Optional[bool] = Field(None, description="Whether the category is active")
72 | order: Optional[int] = Field(None, description="Order of the category")
73 |
74 |
75 | class MoveCatalogItemsParams(BaseModel):
76 | """Parameters for moving catalog items between categories."""
77 |
78 | item_ids: List[str] = Field(..., description="List of catalog item IDs to move")
79 | target_category_id: str = Field(..., description="Target category ID to move items to")
80 |
81 |
82 | def list_catalog_items(
83 | config: ServerConfig,
84 | auth_manager: AuthManager,
85 | params: ListCatalogItemsParams,
86 | ) -> Dict[str, Any]:
87 | """
88 | List service catalog items from ServiceNow.
89 |
90 | Args:
91 | config: Server configuration
92 | auth_manager: Authentication manager
93 | params: Parameters for listing catalog items
94 |
95 | Returns:
96 | Dictionary containing catalog items and metadata
97 | """
98 | logger.info("Listing service catalog items")
99 |
100 | # Build the API URL
101 | url = f"{config.instance_url}/api/now/table/sc_cat_item"
102 |
103 | # Prepare query parameters
104 | query_params = {
105 | "sysparm_limit": params.limit,
106 | "sysparm_offset": params.offset,
107 | "sysparm_display_value": "true",
108 | "sysparm_exclude_reference_link": "true",
109 | }
110 |
111 | # Add filters
112 | filters = []
113 | if params.active:
114 | filters.append("active=true")
115 | if params.category:
116 | filters.append(f"category={params.category}")
117 | if params.query:
118 | filters.append(f"short_descriptionLIKE{params.query}^ORnameLIKE{params.query}")
119 |
120 | if filters:
121 | query_params["sysparm_query"] = "^".join(filters)
122 |
123 | # Make the API request
124 | headers = auth_manager.get_headers()
125 | headers["Accept"] = "application/json"
126 |
127 | try:
128 | response = requests.get(url, headers=headers, params=query_params)
129 | response.raise_for_status()
130 |
131 | # Process the response
132 | result = response.json()
133 | items = result.get("result", [])
134 |
135 | # Format the response
136 | formatted_items = []
137 | for item in items:
138 | formatted_items.append({
139 | "sys_id": item.get("sys_id", ""),
140 | "name": item.get("name", ""),
141 | "short_description": item.get("short_description", ""),
142 | "category": item.get("category", ""),
143 | "price": item.get("price", ""),
144 | "picture": item.get("picture", ""),
145 | "active": item.get("active", ""),
146 | "order": item.get("order", ""),
147 | })
148 |
149 | return {
150 | "success": True,
151 | "message": f"Retrieved {len(formatted_items)} catalog items",
152 | "items": formatted_items,
153 | "total": len(formatted_items),
154 | "limit": params.limit,
155 | "offset": params.offset,
156 | }
157 |
158 | except requests.exceptions.RequestException as e:
159 | logger.error(f"Error listing catalog items: {str(e)}")
160 | return {
161 | "success": False,
162 | "message": f"Error listing catalog items: {str(e)}",
163 | "items": [],
164 | "total": 0,
165 | "limit": params.limit,
166 | "offset": params.offset,
167 | }
168 |
169 |
170 | def get_catalog_item(
171 | config: ServerConfig,
172 | auth_manager: AuthManager,
173 | params: GetCatalogItemParams,
174 | ) -> CatalogResponse:
175 | """
176 | Get a specific service catalog item from ServiceNow.
177 |
178 | Args:
179 | config: Server configuration
180 | auth_manager: Authentication manager
181 | params: Parameters for getting a catalog item
182 |
183 | Returns:
184 | Response containing the catalog item details
185 | """
186 | logger.info(f"Getting service catalog item: {params.item_id}")
187 |
188 | # Build the API URL
189 | url = f"{config.instance_url}/api/now/table/sc_cat_item/{params.item_id}"
190 |
191 | # Prepare query parameters
192 | query_params = {
193 | "sysparm_display_value": "true",
194 | "sysparm_exclude_reference_link": "true",
195 | }
196 |
197 | # Make the API request
198 | headers = auth_manager.get_headers()
199 | headers["Accept"] = "application/json"
200 |
201 | try:
202 | response = requests.get(url, headers=headers, params=query_params)
203 | response.raise_for_status()
204 |
205 | # Process the response
206 | result = response.json()
207 | item = result.get("result", {})
208 |
209 | if not item:
210 | return CatalogResponse(
211 | success=False,
212 | message=f"Catalog item not found: {params.item_id}",
213 | data=None,
214 | )
215 |
216 | # Format the response
217 | formatted_item = {
218 | "sys_id": item.get("sys_id", ""),
219 | "name": item.get("name", ""),
220 | "short_description": item.get("short_description", ""),
221 | "description": item.get("description", ""),
222 | "category": item.get("category", ""),
223 | "price": item.get("price", ""),
224 | "picture": item.get("picture", ""),
225 | "active": item.get("active", ""),
226 | "order": item.get("order", ""),
227 | "delivery_time": item.get("delivery_time", ""),
228 | "availability": item.get("availability", ""),
229 | "variables": get_catalog_item_variables(config, auth_manager, params.item_id),
230 | }
231 |
232 | return CatalogResponse(
233 | success=True,
234 | message=f"Retrieved catalog item: {item.get('name', '')}",
235 | data=formatted_item,
236 | )
237 |
238 | except requests.exceptions.RequestException as e:
239 | logger.error(f"Error getting catalog item: {str(e)}")
240 | return CatalogResponse(
241 | success=False,
242 | message=f"Error getting catalog item: {str(e)}",
243 | data=None,
244 | )
245 |
246 |
247 | def get_catalog_item_variables(
248 | config: ServerConfig,
249 | auth_manager: AuthManager,
250 | item_id: str,
251 | ) -> List[Dict[str, Any]]:
252 | """
253 | Get variables for a specific service catalog item.
254 |
255 | Args:
256 | config: Server configuration
257 | auth_manager: Authentication manager
258 | item_id: Catalog item ID or sys_id
259 |
260 | Returns:
261 | List of variables for the catalog item
262 | """
263 | logger.info(f"Getting variables for catalog item: {item_id}")
264 |
265 | # Build the API URL
266 | url = f"{config.instance_url}/api/now/table/item_option_new"
267 |
268 | # Prepare query parameters
269 | query_params = {
270 | "sysparm_query": f"cat_item={item_id}^ORDERBYorder",
271 | "sysparm_display_value": "true",
272 | "sysparm_exclude_reference_link": "true",
273 | }
274 |
275 | # Make the API request
276 | headers = auth_manager.get_headers()
277 | headers["Accept"] = "application/json"
278 |
279 | try:
280 | response = requests.get(url, headers=headers, params=query_params)
281 | response.raise_for_status()
282 |
283 | # Process the response
284 | result = response.json()
285 | variables = result.get("result", [])
286 |
287 | # Format the response
288 | formatted_variables = []
289 | for variable in variables:
290 | formatted_variables.append({
291 | "sys_id": variable.get("sys_id", ""),
292 | "name": variable.get("name", ""),
293 | "label": variable.get("question_text", ""),
294 | "type": variable.get("type", ""),
295 | "mandatory": variable.get("mandatory", ""),
296 | "default_value": variable.get("default_value", ""),
297 | "help_text": variable.get("help_text", ""),
298 | "order": variable.get("order", ""),
299 | })
300 |
301 | return formatted_variables
302 |
303 | except requests.exceptions.RequestException as e:
304 | logger.error(f"Error getting catalog item variables: {str(e)}")
305 | return []
306 |
307 |
308 | def list_catalog_categories(
309 | config: ServerConfig,
310 | auth_manager: AuthManager,
311 | params: ListCatalogCategoriesParams,
312 | ) -> Dict[str, Any]:
313 | """
314 | List service catalog categories from ServiceNow.
315 |
316 | Args:
317 | config: Server configuration
318 | auth_manager: Authentication manager
319 | params: Parameters for listing catalog categories
320 |
321 | Returns:
322 | Dictionary containing catalog categories and metadata
323 | """
324 | logger.info("Listing service catalog categories")
325 |
326 | # Build the API URL
327 | url = f"{config.instance_url}/api/now/table/sc_category"
328 |
329 | # Prepare query parameters
330 | query_params = {
331 | "sysparm_limit": params.limit,
332 | "sysparm_offset": params.offset,
333 | "sysparm_display_value": "true",
334 | "sysparm_exclude_reference_link": "true",
335 | }
336 |
337 | # Add filters
338 | filters = []
339 | if params.active:
340 | filters.append("active=true")
341 | if params.query:
342 | filters.append(f"titleLIKE{params.query}^ORdescriptionLIKE{params.query}")
343 |
344 | if filters:
345 | query_params["sysparm_query"] = "^".join(filters)
346 |
347 | # Make the API request
348 | headers = auth_manager.get_headers()
349 | headers["Accept"] = "application/json"
350 |
351 | try:
352 | response = requests.get(url, headers=headers, params=query_params)
353 | response.raise_for_status()
354 |
355 | # Process the response
356 | result = response.json()
357 | categories = result.get("result", [])
358 |
359 | # Format the response
360 | formatted_categories = []
361 | for category in categories:
362 | formatted_categories.append({
363 | "sys_id": category.get("sys_id", ""),
364 | "title": category.get("title", ""),
365 | "description": category.get("description", ""),
366 | "parent": category.get("parent", ""),
367 | "icon": category.get("icon", ""),
368 | "active": category.get("active", ""),
369 | "order": category.get("order", ""),
370 | })
371 |
372 | return {
373 | "success": True,
374 | "message": f"Retrieved {len(formatted_categories)} catalog categories",
375 | "categories": formatted_categories,
376 | "total": len(formatted_categories),
377 | "limit": params.limit,
378 | "offset": params.offset,
379 | }
380 |
381 | except requests.exceptions.RequestException as e:
382 | logger.error(f"Error listing catalog categories: {str(e)}")
383 | return {
384 | "success": False,
385 | "message": f"Error listing catalog categories: {str(e)}",
386 | "categories": [],
387 | "total": 0,
388 | "limit": params.limit,
389 | "offset": params.offset,
390 | }
391 |
392 |
393 | def create_catalog_category(
394 | config: ServerConfig,
395 | auth_manager: AuthManager,
396 | params: CreateCatalogCategoryParams,
397 | ) -> CatalogResponse:
398 | """
399 | Create a new service catalog category in ServiceNow.
400 |
401 | Args:
402 | config: Server configuration
403 | auth_manager: Authentication manager
404 | params: Parameters for creating a catalog category
405 |
406 | Returns:
407 | Response containing the result of the operation
408 | """
409 | logger.info("Creating new service catalog category")
410 |
411 | # Build the API URL
412 | url = f"{config.instance_url}/api/now/table/sc_category"
413 |
414 | # Prepare request body
415 | body = {
416 | "title": params.title,
417 | }
418 |
419 | if params.description is not None:
420 | body["description"] = params.description
421 | if params.parent is not None:
422 | body["parent"] = params.parent
423 | if params.icon is not None:
424 | body["icon"] = params.icon
425 | if params.active is not None:
426 | body["active"] = str(params.active).lower()
427 | if params.order is not None:
428 | body["order"] = str(params.order)
429 |
430 | # Make the API request
431 | headers = auth_manager.get_headers()
432 | headers["Accept"] = "application/json"
433 | headers["Content-Type"] = "application/json"
434 |
435 | try:
436 | response = requests.post(url, headers=headers, json=body)
437 | response.raise_for_status()
438 |
439 | # Process the response
440 | result = response.json()
441 | category = result.get("result", {})
442 |
443 | # Format the response
444 | formatted_category = {
445 | "sys_id": category.get("sys_id", ""),
446 | "title": category.get("title", ""),
447 | "description": category.get("description", ""),
448 | "parent": category.get("parent", ""),
449 | "icon": category.get("icon", ""),
450 | "active": category.get("active", ""),
451 | "order": category.get("order", ""),
452 | }
453 |
454 | return CatalogResponse(
455 | success=True,
456 | message=f"Created catalog category: {params.title}",
457 | data=formatted_category,
458 | )
459 |
460 | except requests.exceptions.RequestException as e:
461 | logger.error(f"Error creating catalog category: {str(e)}")
462 | return CatalogResponse(
463 | success=False,
464 | message=f"Error creating catalog category: {str(e)}",
465 | data=None,
466 | )
467 |
468 |
469 | def update_catalog_category(
470 | config: ServerConfig,
471 | auth_manager: AuthManager,
472 | params: UpdateCatalogCategoryParams,
473 | ) -> CatalogResponse:
474 | """
475 | Update an existing service catalog category in ServiceNow.
476 |
477 | Args:
478 | config: Server configuration
479 | auth_manager: Authentication manager
480 | params: Parameters for updating a catalog category
481 |
482 | Returns:
483 | Response containing the result of the operation
484 | """
485 | logger.info(f"Updating service catalog category: {params.category_id}")
486 |
487 | # Build the API URL
488 | url = f"{config.instance_url}/api/now/table/sc_category/{params.category_id}"
489 |
490 | # Prepare request body with only the provided parameters
491 | body = {}
492 | if params.title is not None:
493 | body["title"] = params.title
494 | if params.description is not None:
495 | body["description"] = params.description
496 | if params.parent is not None:
497 | body["parent"] = params.parent
498 | if params.icon is not None:
499 | body["icon"] = params.icon
500 | if params.active is not None:
501 | body["active"] = str(params.active).lower()
502 | if params.order is not None:
503 | body["order"] = str(params.order)
504 |
505 | # Make the API request
506 | headers = auth_manager.get_headers()
507 | headers["Accept"] = "application/json"
508 | headers["Content-Type"] = "application/json"
509 |
510 | try:
511 | response = requests.patch(url, headers=headers, json=body)
512 | response.raise_for_status()
513 |
514 | # Process the response
515 | result = response.json()
516 | category = result.get("result", {})
517 |
518 | # Format the response
519 | formatted_category = {
520 | "sys_id": category.get("sys_id", ""),
521 | "title": category.get("title", ""),
522 | "description": category.get("description", ""),
523 | "parent": category.get("parent", ""),
524 | "icon": category.get("icon", ""),
525 | "active": category.get("active", ""),
526 | "order": category.get("order", ""),
527 | }
528 |
529 | return CatalogResponse(
530 | success=True,
531 | message=f"Updated catalog category: {params.category_id}",
532 | data=formatted_category,
533 | )
534 |
535 | except requests.exceptions.RequestException as e:
536 | logger.error(f"Error updating catalog category: {str(e)}")
537 | return CatalogResponse(
538 | success=False,
539 | message=f"Error updating catalog category: {str(e)}",
540 | data=None,
541 | )
542 |
543 |
544 | def move_catalog_items(
545 | config: ServerConfig,
546 | auth_manager: AuthManager,
547 | params: MoveCatalogItemsParams,
548 | ) -> CatalogResponse:
549 | """
550 | Move catalog items to a different category.
551 |
552 | Args:
553 | config: Server configuration
554 | auth_manager: Authentication manager
555 | params: Parameters for moving catalog items
556 |
557 | Returns:
558 | Response containing the result of the operation
559 | """
560 | logger.info(f"Moving {len(params.item_ids)} catalog items to category: {params.target_category_id}")
561 |
562 | # Build the API URL
563 | url = f"{config.instance_url}/api/now/table/sc_cat_item"
564 |
565 | # Make the API request for each item
566 | headers = auth_manager.get_headers()
567 | headers["Accept"] = "application/json"
568 | headers["Content-Type"] = "application/json"
569 |
570 | success_count = 0
571 | failed_items = []
572 |
573 | try:
574 | for item_id in params.item_ids:
575 | item_url = f"{url}/{item_id}"
576 | body = {
577 | "category": params.target_category_id
578 | }
579 |
580 | try:
581 | response = requests.patch(item_url, headers=headers, json=body)
582 | response.raise_for_status()
583 | success_count += 1
584 | except requests.exceptions.RequestException as e:
585 | logger.error(f"Error moving catalog item {item_id}: {str(e)}")
586 | failed_items.append({"item_id": item_id, "error": str(e)})
587 |
588 | # Prepare the response
589 | if success_count == len(params.item_ids):
590 | return CatalogResponse(
591 | success=True,
592 | message=f"Successfully moved {success_count} catalog items to category {params.target_category_id}",
593 | data={"moved_items_count": success_count},
594 | )
595 | elif success_count > 0:
596 | return CatalogResponse(
597 | success=True,
598 | message=f"Partially moved catalog items. {success_count} succeeded, {len(failed_items)} failed.",
599 | data={
600 | "moved_items_count": success_count,
601 | "failed_items": failed_items,
602 | },
603 | )
604 | else:
605 | return CatalogResponse(
606 | success=False,
607 | message="Failed to move any catalog items",
608 | data={"failed_items": failed_items},
609 | )
610 |
611 | except Exception as e:
612 | logger.error(f"Error moving catalog items: {str(e)}")
613 | return CatalogResponse(
614 | success=False,
615 | message=f"Error moving catalog items: {str(e)}",
616 | data=None,
617 | )
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/incident_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Incident tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing incidents in ServiceNow.
5 | """
6 |
7 | import logging
8 | from typing import Optional, List
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 CreateIncidentParams(BaseModel):
20 | """Parameters for creating an incident."""
21 |
22 | short_description: str = Field(..., description="Short description of the incident")
23 | description: Optional[str] = Field(None, description="Detailed description of the incident")
24 | caller_id: Optional[str] = Field(None, description="User who reported the incident")
25 | category: Optional[str] = Field(None, description="Category of the incident")
26 | subcategory: Optional[str] = Field(None, description="Subcategory of the incident")
27 | priority: Optional[str] = Field(None, description="Priority of the incident")
28 | impact: Optional[str] = Field(None, description="Impact of the incident")
29 | urgency: Optional[str] = Field(None, description="Urgency of the incident")
30 | assigned_to: Optional[str] = Field(None, description="User assigned to the incident")
31 | assignment_group: Optional[str] = Field(None, description="Group assigned to the incident")
32 |
33 |
34 | class UpdateIncidentParams(BaseModel):
35 | """Parameters for updating an incident."""
36 |
37 | incident_id: str = Field(..., description="Incident ID or sys_id")
38 | short_description: Optional[str] = Field(None, description="Short description of the incident")
39 | description: Optional[str] = Field(None, description="Detailed description of the incident")
40 | state: Optional[str] = Field(None, description="State of the incident")
41 | category: Optional[str] = Field(None, description="Category of the incident")
42 | subcategory: Optional[str] = Field(None, description="Subcategory of the incident")
43 | priority: Optional[str] = Field(None, description="Priority of the incident")
44 | impact: Optional[str] = Field(None, description="Impact of the incident")
45 | urgency: Optional[str] = Field(None, description="Urgency of the incident")
46 | assigned_to: Optional[str] = Field(None, description="User assigned to the incident")
47 | assignment_group: Optional[str] = Field(None, description="Group assigned to the incident")
48 | work_notes: Optional[str] = Field(None, description="Work notes to add to the incident")
49 | close_notes: Optional[str] = Field(None, description="Close notes to add to the incident")
50 | close_code: Optional[str] = Field(None, description="Close code for the incident")
51 |
52 |
53 | class AddCommentParams(BaseModel):
54 | """Parameters for adding a comment to an incident."""
55 |
56 | incident_id: str = Field(..., description="Incident ID or sys_id")
57 | comment: str = Field(..., description="Comment to add to the incident")
58 | is_work_note: bool = Field(False, description="Whether the comment is a work note")
59 |
60 |
61 | class ResolveIncidentParams(BaseModel):
62 | """Parameters for resolving an incident."""
63 |
64 | incident_id: str = Field(..., description="Incident ID or sys_id")
65 | resolution_code: str = Field(..., description="Resolution code for the incident")
66 | resolution_notes: str = Field(..., description="Resolution notes for the incident")
67 |
68 |
69 | class ListIncidentsParams(BaseModel):
70 | """Parameters for listing incidents."""
71 |
72 | limit: int = Field(10, description="Maximum number of incidents to return")
73 | offset: int = Field(0, description="Offset for pagination")
74 | state: Optional[str] = Field(None, description="Filter by incident state")
75 | assigned_to: Optional[str] = Field(None, description="Filter by assigned user")
76 | category: Optional[str] = Field(None, description="Filter by category")
77 | query: Optional[str] = Field(None, description="Search query for incidents")
78 |
79 |
80 | class GetIncidentByNumberParams(BaseModel):
81 | """Parameters for fetching an incident by its number."""
82 |
83 | incident_number: str = Field(..., description="The number of the incident to fetch")
84 |
85 |
86 | class IncidentResponse(BaseModel):
87 | """Response from incident operations."""
88 |
89 | success: bool = Field(..., description="Whether the operation was successful")
90 | message: str = Field(..., description="Message describing the result")
91 | incident_id: Optional[str] = Field(None, description="ID of the affected incident")
92 | incident_number: Optional[str] = Field(None, description="Number of the affected incident")
93 |
94 |
95 | def create_incident(
96 | config: ServerConfig,
97 | auth_manager: AuthManager,
98 | params: CreateIncidentParams,
99 | ) -> IncidentResponse:
100 | """
101 | Create a new incident in ServiceNow.
102 |
103 | Args:
104 | config: Server configuration.
105 | auth_manager: Authentication manager.
106 | params: Parameters for creating the incident.
107 |
108 | Returns:
109 | Response with the created incident details.
110 | """
111 | api_url = f"{config.api_url}/table/incident"
112 |
113 | # Build request data
114 | data = {
115 | "short_description": params.short_description,
116 | }
117 |
118 | if params.description:
119 | data["description"] = params.description
120 | if params.caller_id:
121 | data["caller_id"] = params.caller_id
122 | if params.category:
123 | data["category"] = params.category
124 | if params.subcategory:
125 | data["subcategory"] = params.subcategory
126 | if params.priority:
127 | data["priority"] = params.priority
128 | if params.impact:
129 | data["impact"] = params.impact
130 | if params.urgency:
131 | data["urgency"] = params.urgency
132 | if params.assigned_to:
133 | data["assigned_to"] = params.assigned_to
134 | if params.assignment_group:
135 | data["assignment_group"] = params.assignment_group
136 |
137 | # Make request
138 | try:
139 | response = requests.post(
140 | api_url,
141 | json=data,
142 | headers=auth_manager.get_headers(),
143 | timeout=config.timeout,
144 | )
145 | response.raise_for_status()
146 |
147 | result = response.json().get("result", {})
148 |
149 | return IncidentResponse(
150 | success=True,
151 | message="Incident created successfully",
152 | incident_id=result.get("sys_id"),
153 | incident_number=result.get("number"),
154 | )
155 |
156 | except requests.RequestException as e:
157 | logger.error(f"Failed to create incident: {e}")
158 | return IncidentResponse(
159 | success=False,
160 | message=f"Failed to create incident: {str(e)}",
161 | )
162 |
163 |
164 | def update_incident(
165 | config: ServerConfig,
166 | auth_manager: AuthManager,
167 | params: UpdateIncidentParams,
168 | ) -> IncidentResponse:
169 | """
170 | Update an existing incident in ServiceNow.
171 |
172 | Args:
173 | config: Server configuration.
174 | auth_manager: Authentication manager.
175 | params: Parameters for updating the incident.
176 |
177 | Returns:
178 | Response with the updated incident details.
179 | """
180 | # Determine if incident_id is a number or sys_id
181 | incident_id = params.incident_id
182 | if len(incident_id) == 32 and all(c in "0123456789abcdef" for c in incident_id):
183 | # This is likely a sys_id
184 | api_url = f"{config.api_url}/table/incident/{incident_id}"
185 | else:
186 | # This is likely an incident number
187 | # First, we need to get the sys_id
188 | try:
189 | query_url = f"{config.api_url}/table/incident"
190 | query_params = {
191 | "sysparm_query": f"number={incident_id}",
192 | "sysparm_limit": 1,
193 | }
194 |
195 | response = requests.get(
196 | query_url,
197 | params=query_params,
198 | headers=auth_manager.get_headers(),
199 | timeout=config.timeout,
200 | )
201 | response.raise_for_status()
202 |
203 | result = response.json().get("result", [])
204 | if not result:
205 | return IncidentResponse(
206 | success=False,
207 | message=f"Incident not found: {incident_id}",
208 | )
209 |
210 | incident_id = result[0].get("sys_id")
211 | api_url = f"{config.api_url}/table/incident/{incident_id}"
212 |
213 | except requests.RequestException as e:
214 | logger.error(f"Failed to find incident: {e}")
215 | return IncidentResponse(
216 | success=False,
217 | message=f"Failed to find incident: {str(e)}",
218 | )
219 |
220 | # Build request data
221 | data = {}
222 |
223 | if params.short_description:
224 | data["short_description"] = params.short_description
225 | if params.description:
226 | data["description"] = params.description
227 | if params.state:
228 | data["state"] = params.state
229 | if params.category:
230 | data["category"] = params.category
231 | if params.subcategory:
232 | data["subcategory"] = params.subcategory
233 | if params.priority:
234 | data["priority"] = params.priority
235 | if params.impact:
236 | data["impact"] = params.impact
237 | if params.urgency:
238 | data["urgency"] = params.urgency
239 | if params.assigned_to:
240 | data["assigned_to"] = params.assigned_to
241 | if params.assignment_group:
242 | data["assignment_group"] = params.assignment_group
243 | if params.work_notes:
244 | data["work_notes"] = params.work_notes
245 | if params.close_notes:
246 | data["close_notes"] = params.close_notes
247 | if params.close_code:
248 | data["close_code"] = params.close_code
249 |
250 | # Make request
251 | try:
252 | response = requests.put(
253 | api_url,
254 | json=data,
255 | headers=auth_manager.get_headers(),
256 | timeout=config.timeout,
257 | )
258 | response.raise_for_status()
259 |
260 | result = response.json().get("result", {})
261 |
262 | return IncidentResponse(
263 | success=True,
264 | message="Incident updated successfully",
265 | incident_id=result.get("sys_id"),
266 | incident_number=result.get("number"),
267 | )
268 |
269 | except requests.RequestException as e:
270 | logger.error(f"Failed to update incident: {e}")
271 | return IncidentResponse(
272 | success=False,
273 | message=f"Failed to update incident: {str(e)}",
274 | )
275 |
276 |
277 | def add_comment(
278 | config: ServerConfig,
279 | auth_manager: AuthManager,
280 | params: AddCommentParams,
281 | ) -> IncidentResponse:
282 | """
283 | Add a comment to an incident in ServiceNow.
284 |
285 | Args:
286 | config: Server configuration.
287 | auth_manager: Authentication manager.
288 | params: Parameters for adding the comment.
289 |
290 | Returns:
291 | Response with the result of the operation.
292 | """
293 | # Determine if incident_id is a number or sys_id
294 | incident_id = params.incident_id
295 | if len(incident_id) == 32 and all(c in "0123456789abcdef" for c in incident_id):
296 | # This is likely a sys_id
297 | api_url = f"{config.api_url}/table/incident/{incident_id}"
298 | else:
299 | # This is likely an incident number
300 | # First, we need to get the sys_id
301 | try:
302 | query_url = f"{config.api_url}/table/incident"
303 | query_params = {
304 | "sysparm_query": f"number={incident_id}",
305 | "sysparm_limit": 1,
306 | }
307 |
308 | response = requests.get(
309 | query_url,
310 | params=query_params,
311 | headers=auth_manager.get_headers(),
312 | timeout=config.timeout,
313 | )
314 | response.raise_for_status()
315 |
316 | result = response.json().get("result", [])
317 | if not result:
318 | return IncidentResponse(
319 | success=False,
320 | message=f"Incident not found: {incident_id}",
321 | )
322 |
323 | incident_id = result[0].get("sys_id")
324 | api_url = f"{config.api_url}/table/incident/{incident_id}"
325 |
326 | except requests.RequestException as e:
327 | logger.error(f"Failed to find incident: {e}")
328 | return IncidentResponse(
329 | success=False,
330 | message=f"Failed to find incident: {str(e)}",
331 | )
332 |
333 | # Build request data
334 | data = {}
335 |
336 | if params.is_work_note:
337 | data["work_notes"] = params.comment
338 | else:
339 | data["comments"] = params.comment
340 |
341 | # Make request
342 | try:
343 | response = requests.put(
344 | api_url,
345 | json=data,
346 | headers=auth_manager.get_headers(),
347 | timeout=config.timeout,
348 | )
349 | response.raise_for_status()
350 |
351 | result = response.json().get("result", {})
352 |
353 | return IncidentResponse(
354 | success=True,
355 | message="Comment added successfully",
356 | incident_id=result.get("sys_id"),
357 | incident_number=result.get("number"),
358 | )
359 |
360 | except requests.RequestException as e:
361 | logger.error(f"Failed to add comment: {e}")
362 | return IncidentResponse(
363 | success=False,
364 | message=f"Failed to add comment: {str(e)}",
365 | )
366 |
367 |
368 | def resolve_incident(
369 | config: ServerConfig,
370 | auth_manager: AuthManager,
371 | params: ResolveIncidentParams,
372 | ) -> IncidentResponse:
373 | """
374 | Resolve an incident in ServiceNow.
375 |
376 | Args:
377 | config: Server configuration.
378 | auth_manager: Authentication manager.
379 | params: Parameters for resolving the incident.
380 |
381 | Returns:
382 | Response with the result of the operation.
383 | """
384 | # Determine if incident_id is a number or sys_id
385 | incident_id = params.incident_id
386 | if len(incident_id) == 32 and all(c in "0123456789abcdef" for c in incident_id):
387 | # This is likely a sys_id
388 | api_url = f"{config.api_url}/table/incident/{incident_id}"
389 | else:
390 | # This is likely an incident number
391 | # First, we need to get the sys_id
392 | try:
393 | query_url = f"{config.api_url}/table/incident"
394 | query_params = {
395 | "sysparm_query": f"number={incident_id}",
396 | "sysparm_limit": 1,
397 | }
398 |
399 | response = requests.get(
400 | query_url,
401 | params=query_params,
402 | headers=auth_manager.get_headers(),
403 | timeout=config.timeout,
404 | )
405 | response.raise_for_status()
406 |
407 | result = response.json().get("result", [])
408 | if not result:
409 | return IncidentResponse(
410 | success=False,
411 | message=f"Incident not found: {incident_id}",
412 | )
413 |
414 | incident_id = result[0].get("sys_id")
415 | api_url = f"{config.api_url}/table/incident/{incident_id}"
416 |
417 | except requests.RequestException as e:
418 | logger.error(f"Failed to find incident: {e}")
419 | return IncidentResponse(
420 | success=False,
421 | message=f"Failed to find incident: {str(e)}",
422 | )
423 |
424 | # Build request data
425 | data = {
426 | "state": "6", # Resolved
427 | "close_code": params.resolution_code,
428 | "close_notes": params.resolution_notes,
429 | "resolved_at": "now",
430 | }
431 |
432 | # Make request
433 | try:
434 | response = requests.put(
435 | api_url,
436 | json=data,
437 | headers=auth_manager.get_headers(),
438 | timeout=config.timeout,
439 | )
440 | response.raise_for_status()
441 |
442 | result = response.json().get("result", {})
443 |
444 | return IncidentResponse(
445 | success=True,
446 | message="Incident resolved successfully",
447 | incident_id=result.get("sys_id"),
448 | incident_number=result.get("number"),
449 | )
450 |
451 | except requests.RequestException as e:
452 | logger.error(f"Failed to resolve incident: {e}")
453 | return IncidentResponse(
454 | success=False,
455 | message=f"Failed to resolve incident: {str(e)}",
456 | )
457 |
458 |
459 | def list_incidents(
460 | config: ServerConfig,
461 | auth_manager: AuthManager,
462 | params: ListIncidentsParams,
463 | ) -> dict:
464 | """
465 | List incidents from ServiceNow.
466 |
467 | Args:
468 | config: Server configuration.
469 | auth_manager: Authentication manager.
470 | params: Parameters for listing incidents.
471 |
472 | Returns:
473 | Dictionary with list of incidents.
474 | """
475 | api_url = f"{config.api_url}/table/incident"
476 |
477 | # Build query parameters
478 | query_params = {
479 | "sysparm_limit": params.limit,
480 | "sysparm_offset": params.offset,
481 | "sysparm_display_value": "true",
482 | "sysparm_exclude_reference_link": "true",
483 | }
484 |
485 | # Add filters
486 | filters = []
487 | if params.state:
488 | filters.append(f"state={params.state}")
489 | if params.assigned_to:
490 | filters.append(f"assigned_to={params.assigned_to}")
491 | if params.category:
492 | filters.append(f"category={params.category}")
493 | if params.query:
494 | filters.append(f"short_descriptionLIKE{params.query}^ORdescriptionLIKE{params.query}")
495 |
496 | if filters:
497 | query_params["sysparm_query"] = "^".join(filters)
498 |
499 | # Make request
500 | try:
501 | response = requests.get(
502 | api_url,
503 | params=query_params,
504 | headers=auth_manager.get_headers(),
505 | timeout=config.timeout,
506 | )
507 | response.raise_for_status()
508 |
509 | data = response.json()
510 | incidents = []
511 |
512 | for incident_data in data.get("result", []):
513 | # Handle assigned_to field which could be a string or a dictionary
514 | assigned_to = incident_data.get("assigned_to")
515 | if isinstance(assigned_to, dict):
516 | assigned_to = assigned_to.get("display_value")
517 |
518 | incident = {
519 | "sys_id": incident_data.get("sys_id"),
520 | "number": incident_data.get("number"),
521 | "short_description": incident_data.get("short_description"),
522 | "description": incident_data.get("description"),
523 | "state": incident_data.get("state"),
524 | "priority": incident_data.get("priority"),
525 | "assigned_to": assigned_to,
526 | "category": incident_data.get("category"),
527 | "subcategory": incident_data.get("subcategory"),
528 | "created_on": incident_data.get("sys_created_on"),
529 | "updated_on": incident_data.get("sys_updated_on"),
530 | }
531 | incidents.append(incident)
532 |
533 | return {
534 | "success": True,
535 | "message": f"Found {len(incidents)} incidents",
536 | "incidents": incidents
537 | }
538 |
539 | except requests.RequestException as e:
540 | logger.error(f"Failed to list incidents: {e}")
541 | return {
542 | "success": False,
543 | "message": f"Failed to list incidents: {str(e)}",
544 | "incidents": []
545 | }
546 |
547 |
548 | def get_incident_by_number(
549 | config: ServerConfig,
550 | auth_manager: AuthManager,
551 | params: GetIncidentByNumberParams,
552 | ) -> dict:
553 | """
554 | Fetch a single incident from ServiceNow by its number.
555 |
556 | Args:
557 | config: Server configuration.
558 | auth_manager: Authentication manager.
559 | params: Parameters for fetching the incident.
560 |
561 | Returns:
562 | Dictionary with the incident details.
563 | """
564 | api_url = f"{config.api_url}/table/incident"
565 |
566 | # Build query parameters
567 | query_params = {
568 | "sysparm_query": f"number={params.incident_number}",
569 | "sysparm_limit": 1,
570 | "sysparm_display_value": "true",
571 | "sysparm_exclude_reference_link": "true",
572 | }
573 |
574 | # Make request
575 | try:
576 | response = requests.get(
577 | api_url,
578 | params=query_params,
579 | headers=auth_manager.get_headers(),
580 | timeout=config.timeout,
581 | )
582 | response.raise_for_status()
583 |
584 | data = response.json()
585 | result = data.get("result", [])
586 |
587 | if not result:
588 | return {
589 | "success": False,
590 | "message": f"Incident not found: {params.incident_number}",
591 | }
592 |
593 | incident_data = result[0]
594 | assigned_to = incident_data.get("assigned_to")
595 | if isinstance(assigned_to, dict):
596 | assigned_to = assigned_to.get("display_value")
597 |
598 | incident = {
599 | "sys_id": incident_data.get("sys_id"),
600 | "number": incident_data.get("number"),
601 | "short_description": incident_data.get("short_description"),
602 | "description": incident_data.get("description"),
603 | "state": incident_data.get("state"),
604 | "priority": incident_data.get("priority"),
605 | "assigned_to": assigned_to,
606 | "category": incident_data.get("category"),
607 | "subcategory": incident_data.get("subcategory"),
608 | "created_on": incident_data.get("sys_created_on"),
609 | "updated_on": incident_data.get("sys_updated_on"),
610 | }
611 |
612 | return {
613 | "success": True,
614 | "message": f"Incident {params.incident_number} found",
615 | "incident": incident,
616 | }
617 |
618 | except requests.RequestException as e:
619 | logger.error(f"Failed to fetch incident: {e}")
620 | return {
621 | "success": False,
622 | "message": f"Failed to fetch incident: {str(e)}",
623 | }
624 |
```
--------------------------------------------------------------------------------
/tests/test_catalog_optimization.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the ServiceNow MCP catalog optimization 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.catalog_optimization import (
12 | OptimizationRecommendationsParams,
13 | UpdateCatalogItemParams,
14 | _get_high_abandonment_items,
15 | _get_inactive_items,
16 | _get_low_usage_items,
17 | _get_poor_description_items,
18 | _get_slow_fulfillment_items,
19 | get_optimization_recommendations,
20 | update_catalog_item,
21 | )
22 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
23 |
24 |
25 | class TestCatalogOptimizationTools(unittest.TestCase):
26 | """Test cases for the catalog optimization tools."""
27 |
28 | def setUp(self):
29 | """Set up test fixtures."""
30 | # Create a mock server config
31 | self.config = ServerConfig(
32 | instance_url="https://example.service-now.com",
33 | auth=AuthConfig(
34 | type=AuthType.BASIC,
35 | basic=BasicAuthConfig(username="admin", password="password"),
36 | ),
37 | )
38 |
39 | # Create a mock auth manager
40 | self.auth_manager = MagicMock(spec=AuthManager)
41 | self.auth_manager.get_headers.return_value = {"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}
42 |
43 | @patch("requests.get")
44 | def test_get_inactive_items(self, mock_get):
45 | """Test getting inactive catalog items."""
46 | # Mock the response from ServiceNow
47 | mock_response = MagicMock()
48 | mock_response.json.return_value = {
49 | "result": [
50 | {
51 | "sys_id": "item1",
52 | "name": "Old Laptop",
53 | "short_description": "Outdated laptop model",
54 | "category": "hardware",
55 | },
56 | {
57 | "sys_id": "item2",
58 | "name": "Legacy Software",
59 | "short_description": "Deprecated software package",
60 | "category": "software",
61 | },
62 | ]
63 | }
64 | mock_get.return_value = mock_response
65 |
66 | # Call the function
67 | result = _get_inactive_items(self.config, self.auth_manager)
68 |
69 | # Verify the results
70 | self.assertEqual(len(result), 2)
71 | self.assertEqual(result[0]["name"], "Old Laptop")
72 | self.assertEqual(result[1]["name"], "Legacy Software")
73 |
74 | # Verify the API call
75 | mock_get.assert_called_once()
76 | args, kwargs = mock_get.call_args
77 | self.assertEqual(kwargs["params"]["sysparm_query"], "active=false")
78 |
79 | @patch("requests.get")
80 | def test_get_inactive_items_with_category(self, mock_get):
81 | """Test getting inactive catalog items filtered by category."""
82 | # Mock the response from ServiceNow
83 | mock_response = MagicMock()
84 | mock_response.json.return_value = {
85 | "result": [
86 | {
87 | "sys_id": "item1",
88 | "name": "Old Laptop",
89 | "short_description": "Outdated laptop model",
90 | "category": "hardware",
91 | },
92 | ]
93 | }
94 | mock_get.return_value = mock_response
95 |
96 | # Call the function with a category filter
97 | result = _get_inactive_items(self.config, self.auth_manager, "hardware")
98 |
99 | # Verify the results
100 | self.assertEqual(len(result), 1)
101 | self.assertEqual(result[0]["name"], "Old Laptop")
102 |
103 | # Verify the API call
104 | mock_get.assert_called_once()
105 | args, kwargs = mock_get.call_args
106 | self.assertEqual(kwargs["params"]["sysparm_query"], "active=false^category=hardware")
107 |
108 | @patch("requests.get")
109 | def test_get_inactive_items_error(self, mock_get):
110 | """Test error handling when getting inactive catalog items."""
111 | # Mock an error response
112 | mock_get.side_effect = requests.exceptions.RequestException("API Error")
113 |
114 | # Call the function
115 | result = _get_inactive_items(self.config, self.auth_manager)
116 |
117 | # Verify the results
118 | self.assertEqual(result, [])
119 |
120 | @patch("requests.get")
121 | @patch("random.sample")
122 | @patch("random.randint")
123 | def test_get_low_usage_items(self, mock_randint, mock_sample, mock_get):
124 | """Test getting catalog items with low usage."""
125 | # Mock the response from ServiceNow
126 | mock_response = MagicMock()
127 | mock_response.json.return_value = {
128 | "result": [
129 | {
130 | "sys_id": "item1",
131 | "name": "Rarely Used Laptop",
132 | "short_description": "Laptop model with low demand",
133 | "category": "hardware",
134 | },
135 | {
136 | "sys_id": "item2",
137 | "name": "Unpopular Software",
138 | "short_description": "Software with few users",
139 | "category": "software",
140 | },
141 | {
142 | "sys_id": "item3",
143 | "name": "Niche Service",
144 | "short_description": "Specialized service with limited audience",
145 | "category": "services",
146 | },
147 | ]
148 | }
149 | mock_get.return_value = mock_response
150 |
151 | # Mock the random sample to return the first two items
152 | mock_sample.return_value = [
153 | {
154 | "sys_id": "item1",
155 | "name": "Rarely Used Laptop",
156 | "short_description": "Laptop model with low demand",
157 | "category": "hardware",
158 | },
159 | {
160 | "sys_id": "item2",
161 | "name": "Unpopular Software",
162 | "short_description": "Software with few users",
163 | "category": "software",
164 | },
165 | ]
166 |
167 | # Mock the random order counts
168 | mock_randint.return_value = 2
169 |
170 | # Call the function
171 | result = _get_low_usage_items(self.config, self.auth_manager)
172 |
173 | # Verify the results
174 | self.assertEqual(len(result), 2)
175 | self.assertEqual(result[0]["name"], "Rarely Used Laptop")
176 | self.assertEqual(result[0]["order_count"], 2)
177 | self.assertEqual(result[1]["name"], "Unpopular Software")
178 | self.assertEqual(result[1]["order_count"], 2)
179 |
180 | # Verify the API call
181 | mock_get.assert_called_once()
182 | args, kwargs = mock_get.call_args
183 | self.assertEqual(kwargs["params"]["sysparm_query"], "active=true")
184 |
185 | def test_high_abandonment_items_format(self):
186 | """Test the expected format of high abandonment items."""
187 | # This test doesn't call the actual function, but verifies the expected format
188 | # of the data that would be returned by the function
189 |
190 | # Example data that would be returned by _get_high_abandonment_items
191 | high_abandonment_items = [
192 | {
193 | "sys_id": "item1",
194 | "name": "Complex Request",
195 | "short_description": "Request with many fields",
196 | "category": "hardware",
197 | "abandonment_rate": 60,
198 | "cart_adds": 30,
199 | "orders": 12,
200 | },
201 | {
202 | "sys_id": "item2",
203 | "name": "Expensive Item",
204 | "short_description": "High-cost item",
205 | "category": "software",
206 | "abandonment_rate": 60,
207 | "cart_adds": 20,
208 | "orders": 8,
209 | },
210 | ]
211 |
212 | # Verify the expected format
213 | self.assertEqual(len(high_abandonment_items), 2)
214 | self.assertEqual(high_abandonment_items[0]["name"], "Complex Request")
215 | self.assertEqual(high_abandonment_items[0]["abandonment_rate"], 60)
216 | self.assertEqual(high_abandonment_items[0]["cart_adds"], 30)
217 | self.assertEqual(high_abandonment_items[0]["orders"], 12)
218 | self.assertEqual(high_abandonment_items[1]["name"], "Expensive Item")
219 | self.assertEqual(high_abandonment_items[1]["abandonment_rate"], 60)
220 | self.assertEqual(high_abandonment_items[1]["cart_adds"], 20)
221 | self.assertEqual(high_abandonment_items[1]["orders"], 8)
222 |
223 | @patch("requests.get")
224 | @patch("random.sample")
225 | @patch("random.uniform")
226 | def test_get_slow_fulfillment_items(self, mock_uniform, mock_sample, mock_get):
227 | """Test getting catalog items with slow fulfillment times."""
228 | # Mock the response from ServiceNow
229 | mock_response = MagicMock()
230 | mock_response.json.return_value = {
231 | "result": [
232 | {
233 | "sys_id": "item1",
234 | "name": "Custom Hardware",
235 | "short_description": "Specialized hardware request",
236 | "category": "hardware",
237 | },
238 | {
239 | "sys_id": "item2",
240 | "name": "Complex Software",
241 | "short_description": "Software with complex installation",
242 | "category": "software",
243 | },
244 | ]
245 | }
246 | mock_get.return_value = mock_response
247 |
248 | # Mock the random sample to return all items
249 | mock_sample.return_value = [
250 | {
251 | "sys_id": "item1",
252 | "name": "Custom Hardware",
253 | "short_description": "Specialized hardware request",
254 | "category": "hardware",
255 | },
256 | {
257 | "sys_id": "item2",
258 | "name": "Complex Software",
259 | "short_description": "Software with complex installation",
260 | "category": "software",
261 | },
262 | ]
263 |
264 | # Mock the random uniform values for fulfillment times
265 | mock_uniform.return_value = 7.5
266 |
267 | # Call the function
268 | result = _get_slow_fulfillment_items(self.config, self.auth_manager)
269 |
270 | # Verify the results
271 | self.assertEqual(len(result), 2)
272 | self.assertEqual(result[0]["name"], "Custom Hardware")
273 | self.assertEqual(result[0]["avg_fulfillment_time"], 7.5)
274 | self.assertEqual(result[0]["avg_fulfillment_time_vs_catalog"], 3.0) # 7.5 / 2.5 = 3.0
275 | self.assertEqual(result[1]["name"], "Complex Software")
276 | self.assertEqual(result[1]["avg_fulfillment_time"], 7.5)
277 | self.assertEqual(result[1]["avg_fulfillment_time_vs_catalog"], 3.0) # 7.5 / 2.5 = 3.0
278 |
279 | @patch("requests.get")
280 | def test_get_poor_description_items(self, mock_get):
281 | """Test getting catalog items with poor description quality."""
282 | # Mock the response from ServiceNow
283 | mock_response = MagicMock()
284 | mock_response.json.return_value = {
285 | "result": [
286 | {
287 | "sys_id": "item1",
288 | "name": "Laptop",
289 | "short_description": "", # Empty description
290 | "category": "hardware",
291 | },
292 | {
293 | "sys_id": "item2",
294 | "name": "Software",
295 | "short_description": "Software package", # Short description
296 | "category": "software",
297 | },
298 | {
299 | "sys_id": "item3",
300 | "name": "Service",
301 | "short_description": "Please click here to request this service", # Instructional language
302 | "category": "services",
303 | },
304 | ]
305 | }
306 | mock_get.return_value = mock_response
307 |
308 | # Call the function
309 | result = _get_poor_description_items(self.config, self.auth_manager)
310 |
311 | # Verify the results
312 | self.assertEqual(len(result), 3)
313 |
314 | # Check the first item (empty description)
315 | self.assertEqual(result[0]["name"], "Laptop")
316 | self.assertEqual(result[0]["description_quality"], 0)
317 | self.assertEqual(result[0]["quality_issues"], ["Missing description"])
318 |
319 | # Check the second item (short description)
320 | self.assertEqual(result[1]["name"], "Software")
321 | self.assertEqual(result[1]["description_quality"], 30)
322 | self.assertEqual(result[1]["quality_issues"], ["Description too short", "Lacks detail"])
323 |
324 | # Check the third item (instructional language)
325 | self.assertEqual(result[2]["name"], "Service")
326 | self.assertEqual(result[2]["description_quality"], 50)
327 | self.assertEqual(result[2]["quality_issues"], ["Uses instructional language instead of descriptive"])
328 |
329 | @patch("servicenow_mcp.tools.catalog_optimization._get_inactive_items")
330 | @patch("servicenow_mcp.tools.catalog_optimization._get_low_usage_items")
331 | @patch("servicenow_mcp.tools.catalog_optimization._get_high_abandonment_items")
332 | @patch("servicenow_mcp.tools.catalog_optimization._get_slow_fulfillment_items")
333 | @patch("servicenow_mcp.tools.catalog_optimization._get_poor_description_items")
334 | def test_get_optimization_recommendations(
335 | self,
336 | mock_poor_desc,
337 | mock_slow_fulfill,
338 | mock_high_abandon,
339 | mock_low_usage,
340 | mock_inactive
341 | ):
342 | """Test getting optimization recommendations."""
343 | # Mock the helper functions to return test data
344 | mock_inactive.return_value = [
345 | {
346 | "sys_id": "item1",
347 | "name": "Old Laptop",
348 | "short_description": "Outdated laptop model",
349 | "category": "hardware",
350 | },
351 | ]
352 |
353 | mock_low_usage.return_value = [
354 | {
355 | "sys_id": "item2",
356 | "name": "Rarely Used Software",
357 | "short_description": "Software with few users",
358 | "category": "software",
359 | "order_count": 2,
360 | },
361 | ]
362 |
363 | mock_high_abandon.return_value = [
364 | {
365 | "sys_id": "item3",
366 | "name": "Complex Request",
367 | "short_description": "Request with many fields",
368 | "category": "hardware",
369 | "abandonment_rate": 60,
370 | "cart_adds": 30,
371 | "orders": 12,
372 | },
373 | ]
374 |
375 | mock_slow_fulfill.return_value = [
376 | {
377 | "sys_id": "item4",
378 | "name": "Custom Hardware",
379 | "short_description": "Specialized hardware request",
380 | "category": "hardware",
381 | "avg_fulfillment_time": 7.5,
382 | "avg_fulfillment_time_vs_catalog": 3.0,
383 | },
384 | ]
385 |
386 | mock_poor_desc.return_value = [
387 | {
388 | "sys_id": "item5",
389 | "name": "Laptop",
390 | "short_description": "",
391 | "category": "hardware",
392 | "description_quality": 0,
393 | "quality_issues": ["Missing description"],
394 | },
395 | ]
396 |
397 | # Create the parameters
398 | params = OptimizationRecommendationsParams(
399 | recommendation_types=[
400 | "inactive_items",
401 | "low_usage",
402 | "high_abandonment",
403 | "slow_fulfillment",
404 | "description_quality"
405 | ]
406 | )
407 |
408 | # Call the function
409 | result = get_optimization_recommendations(self.config, self.auth_manager, params)
410 |
411 | # Verify the results
412 | self.assertTrue(result["success"])
413 | self.assertEqual(len(result["recommendations"]), 5)
414 |
415 | # Check each recommendation type
416 | recommendation_types = [rec["type"] for rec in result["recommendations"]]
417 | self.assertIn("inactive_items", recommendation_types)
418 | self.assertIn("low_usage", recommendation_types)
419 | self.assertIn("high_abandonment", recommendation_types)
420 | self.assertIn("slow_fulfillment", recommendation_types)
421 | self.assertIn("description_quality", recommendation_types)
422 |
423 | # Check that each recommendation has the expected fields
424 | for rec in result["recommendations"]:
425 | self.assertIn("title", rec)
426 | self.assertIn("description", rec)
427 | self.assertIn("items", rec)
428 | self.assertIn("impact", rec)
429 | self.assertIn("effort", rec)
430 | self.assertIn("action", rec)
431 |
432 | @patch("servicenow_mcp.tools.catalog_optimization._get_inactive_items")
433 | @patch("servicenow_mcp.tools.catalog_optimization._get_low_usage_items")
434 | def test_get_optimization_recommendations_filtered(self, mock_low_usage, mock_inactive):
435 | """Test getting filtered optimization recommendations."""
436 | # Mock the helper functions to return test data
437 | mock_inactive.return_value = [
438 | {
439 | "sys_id": "item1",
440 | "name": "Old Laptop",
441 | "short_description": "Outdated laptop model",
442 | "category": "hardware",
443 | },
444 | ]
445 |
446 | mock_low_usage.return_value = [
447 | {
448 | "sys_id": "item2",
449 | "name": "Rarely Used Software",
450 | "short_description": "Software with few users",
451 | "category": "software",
452 | "order_count": 2,
453 | },
454 | ]
455 |
456 | # Create the parameters with only specific recommendation types
457 | params = OptimizationRecommendationsParams(
458 | recommendation_types=["inactive_items", "low_usage"]
459 | )
460 |
461 | # Call the function
462 | result = get_optimization_recommendations(self.config, self.auth_manager, params)
463 |
464 | # Verify the results
465 | self.assertTrue(result["success"])
466 | self.assertEqual(len(result["recommendations"]), 2)
467 |
468 | # Check each recommendation type
469 | recommendation_types = [rec["type"] for rec in result["recommendations"]]
470 | self.assertIn("inactive_items", recommendation_types)
471 | self.assertIn("low_usage", recommendation_types)
472 | self.assertNotIn("high_abandonment", recommendation_types)
473 | self.assertNotIn("slow_fulfillment", recommendation_types)
474 | self.assertNotIn("description_quality", recommendation_types)
475 |
476 | @patch("requests.patch")
477 | def test_update_catalog_item(self, mock_patch):
478 | """Test updating a catalog item."""
479 | # Mock the response from ServiceNow
480 | mock_response = MagicMock()
481 | mock_response.json.return_value = {
482 | "result": {
483 | "sys_id": "item1",
484 | "name": "Laptop",
485 | "short_description": "Updated laptop description",
486 | "description": "Detailed description",
487 | "category": "hardware",
488 | "price": "999.99",
489 | "active": "true",
490 | "order": "100",
491 | }
492 | }
493 | mock_patch.return_value = mock_response
494 |
495 | # Create the parameters
496 | params = UpdateCatalogItemParams(
497 | item_id="item1",
498 | short_description="Updated laptop description",
499 | )
500 |
501 | # Call the function
502 | result = update_catalog_item(self.config, self.auth_manager, params)
503 |
504 | # Verify the results
505 | self.assertTrue(result["success"])
506 | self.assertEqual(result["data"]["short_description"], "Updated laptop description")
507 |
508 | # Verify the API call
509 | mock_patch.assert_called_once()
510 | args, kwargs = mock_patch.call_args
511 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item/item1")
512 | self.assertEqual(kwargs["json"], {"short_description": "Updated laptop description"})
513 |
514 | @patch("requests.patch")
515 | def test_update_catalog_item_multiple_fields(self, mock_patch):
516 | """Test updating multiple fields of a catalog item."""
517 | # Mock the response from ServiceNow
518 | mock_response = MagicMock()
519 | mock_response.json.return_value = {
520 | "result": {
521 | "sys_id": "item1",
522 | "name": "Updated Laptop",
523 | "short_description": "Updated laptop description",
524 | "description": "Detailed description",
525 | "category": "hardware",
526 | "price": "1099.99",
527 | "active": "true",
528 | "order": "100",
529 | }
530 | }
531 | mock_patch.return_value = mock_response
532 |
533 | # Create the parameters with multiple fields
534 | params = UpdateCatalogItemParams(
535 | item_id="item1",
536 | name="Updated Laptop",
537 | short_description="Updated laptop description",
538 | price="1099.99",
539 | )
540 |
541 | # Call the function
542 | result = update_catalog_item(self.config, self.auth_manager, params)
543 |
544 | # Verify the results
545 | self.assertTrue(result["success"])
546 | self.assertEqual(result["data"]["name"], "Updated Laptop")
547 | self.assertEqual(result["data"]["short_description"], "Updated laptop description")
548 | self.assertEqual(result["data"]["price"], "1099.99")
549 |
550 | # Verify the API call
551 | mock_patch.assert_called_once()
552 | args, kwargs = mock_patch.call_args
553 | self.assertEqual(args[0], "https://example.service-now.com/api/now/table/sc_cat_item/item1")
554 | self.assertEqual(kwargs["json"], {
555 | "name": "Updated Laptop",
556 | "short_description": "Updated laptop description",
557 | "price": "1099.99",
558 | })
559 |
560 | @patch("requests.patch")
561 | def test_update_catalog_item_error(self, mock_patch):
562 | """Test error handling when updating a catalog item."""
563 | # Mock an error response
564 | mock_patch.side_effect = requests.exceptions.RequestException("API Error")
565 |
566 | # Create the parameters
567 | params = UpdateCatalogItemParams(
568 | item_id="item1",
569 | short_description="Updated laptop description",
570 | )
571 |
572 | # Call the function
573 | result = update_catalog_item(self.config, self.auth_manager, params)
574 |
575 | # Verify the results
576 | self.assertFalse(result["success"])
577 | self.assertIn("Error updating catalog item", result["message"])
578 | self.assertIsNone(result["data"])
579 |
580 |
581 | if __name__ == "__main__":
582 | unittest.main()
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/story_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Story management tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing stories in ServiceNow.
5 | """
6 |
7 | import logging
8 | from datetime import datetime
9 | from typing import Any, Dict, List, Optional, Type, TypeVar
10 |
11 | import requests
12 | from pydantic import BaseModel, Field
13 |
14 | from servicenow_mcp.auth.auth_manager import AuthManager
15 | from servicenow_mcp.utils.config import ServerConfig
16 |
17 | logger = logging.getLogger(__name__)
18 |
19 | # Type variable for Pydantic models
20 | T = TypeVar('T', bound=BaseModel)
21 |
22 | class CreateStoryParams(BaseModel):
23 | """Parameters for creating a story."""
24 |
25 | short_description: str = Field(..., description="Short description of the story")
26 | acceptance_criteria: str = Field(..., description="Acceptance criteria for the story")
27 | description: Optional[str] = Field(None, description="Detailed description of the story")
28 | state: Optional[str] = Field(None, description="State of story (-6 is Draft,-7 is Ready for Testing,-8 is Testing,1 is Ready, 2 is Work in progress, 3 is Complete, 4 is Cancelled)")
29 | assignment_group: Optional[str] = Field(None, description="Group assigned to the story")
30 | story_points: Optional[int] = Field(10, description="Points value for the story")
31 | assigned_to: Optional[str] = Field(None, description="User assigned to the story")
32 | epic: Optional[str] = Field(None, description="Epic that the story belongs to. It requires the System ID of the epic.")
33 | project: Optional[str] = Field(None, description="Project that the story belongs to. It requires the System ID of the project.")
34 | work_notes: Optional[str] = Field(None, description="Work notes to add to the story. Used for adding notes and comments to a story")
35 |
36 | class UpdateStoryParams(BaseModel):
37 | """Parameters for updating a story."""
38 |
39 | story_id: str = Field(..., description="Story IDNumber or sys_id. You will need to fetch the story to get the sys_id if you only have the story number")
40 | short_description: Optional[str] = Field(None, description="Short description of the story")
41 | acceptance_criteria: Optional[str] = Field(None, description="Acceptance criteria for the story")
42 | description: Optional[str] = Field(None, description="Detailed description of the story")
43 | state: Optional[str] = Field(None, description="State of story (-6 is Draft,-7 is Ready for Testing,-8 is Testing,1 is Ready, 2 is Work in progress, 3 is Complete, 4 is Cancelled)")
44 | assignment_group: Optional[str] = Field(None, description="Group assigned to the story")
45 | story_points: Optional[int] = Field(None, description="Points value for the story")
46 | assigned_to: Optional[str] = Field(None, description="User assigned to the story")
47 | epic: Optional[str] = Field(None, description="Epic that the story belongs to. It requires the System ID of the epic.")
48 | project: Optional[str] = Field(None, description="Project that the story belongs to. It requires the System ID of the project.")
49 | work_notes: Optional[str] = Field(None, description="Work notes to add to the story. Used for adding notes and comments to a story")
50 |
51 | class ListStoriesParams(BaseModel):
52 | """Parameters for listing stories."""
53 |
54 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
55 | offset: Optional[int] = Field(0, description="Offset to start from")
56 | state: Optional[str] = Field(None, description="Filter by state")
57 | assignment_group: Optional[str] = Field(None, description="Filter by assignment group")
58 | timeframe: Optional[str] = Field(None, description="Filter by timeframe (upcoming, in-progress, completed)")
59 | query: Optional[str] = Field(None, description="Additional query string")
60 |
61 | class ListStoryDependenciesParams(BaseModel):
62 | """Parameters for listing story dependencies."""
63 |
64 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
65 | offset: Optional[int] = Field(0, description="Offset to start from")
66 | query: Optional[str] = Field(None, description="Additional query string")
67 | dependent_story: Optional[str] = Field(None, description="Sys_id of the dependent story is required")
68 | prerequisite_story: Optional[str] = Field(None, description="Sys_id that this story depends on is required")
69 |
70 | class CreateStoryDependencyParams(BaseModel):
71 | """Parameters for creating a story dependency."""
72 |
73 | dependent_story: str = Field(..., description="Sys_id of the dependent story is required")
74 | prerequisite_story: str = Field(..., description="Sys_id that this story depends on is required")
75 |
76 | class DeleteStoryDependencyParams(BaseModel):
77 | """Parameters for deleting a story dependency."""
78 |
79 | dependency_id: str = Field(..., description="Sys_id of the dependency is required")
80 |
81 | def _unwrap_and_validate_params(params: Any, model_class: Type[T], required_fields: List[str] = None) -> Dict[str, Any]:
82 | """
83 | Helper function to unwrap and validate parameters.
84 |
85 | Args:
86 | params: The parameters to unwrap and validate.
87 | model_class: The Pydantic model class to validate against.
88 | required_fields: List of required field names.
89 |
90 | Returns:
91 | A tuple of (success, result) where result is either the validated parameters or an error message.
92 | """
93 | # Handle case where params might be wrapped in another dictionary
94 | if isinstance(params, dict) and len(params) == 1 and "params" in params and isinstance(params["params"], dict):
95 | logger.warning("Detected params wrapped in a 'params' key. Unwrapping...")
96 | params = params["params"]
97 |
98 | # Handle case where params might be a Pydantic model object
99 | if not isinstance(params, dict):
100 | try:
101 | # Try to convert to dict if it's a Pydantic model
102 | logger.warning("Params is not a dictionary. Attempting to convert...")
103 | params = params.dict() if hasattr(params, "dict") else dict(params)
104 | except Exception as e:
105 | logger.error(f"Failed to convert params to dictionary: {e}")
106 | return {
107 | "success": False,
108 | "message": f"Invalid parameters format. Expected a dictionary, got {type(params).__name__}",
109 | }
110 |
111 | # Validate required parameters are present
112 | if required_fields:
113 | for field in required_fields:
114 | if field not in params:
115 | return {
116 | "success": False,
117 | "message": f"Missing required parameter '{field}'",
118 | }
119 |
120 | try:
121 | # Validate parameters against the model
122 | validated_params = model_class(**params)
123 | return {
124 | "success": True,
125 | "params": validated_params,
126 | }
127 | except Exception as e:
128 | logger.error(f"Error validating parameters: {e}")
129 | return {
130 | "success": False,
131 | "message": f"Error validating parameters: {str(e)}",
132 | }
133 |
134 |
135 | def _get_instance_url(auth_manager: AuthManager, server_config: ServerConfig) -> Optional[str]:
136 | """
137 | Helper function to get the instance URL from either server_config or auth_manager.
138 |
139 | Args:
140 | auth_manager: The authentication manager.
141 | server_config: The server configuration.
142 |
143 | Returns:
144 | The instance URL if found, None otherwise.
145 | """
146 | if hasattr(server_config, 'instance_url'):
147 | return server_config.instance_url
148 | elif hasattr(auth_manager, 'instance_url'):
149 | return auth_manager.instance_url
150 | else:
151 | logger.error("Cannot find instance_url in either server_config or auth_manager")
152 | return None
153 |
154 |
155 | def _get_headers(auth_manager: Any, server_config: Any) -> Optional[Dict[str, str]]:
156 | """
157 | Helper function to get headers from either auth_manager or server_config.
158 |
159 | Args:
160 | auth_manager: The authentication manager or object passed as auth_manager.
161 | server_config: The server configuration or object passed as server_config.
162 |
163 | Returns:
164 | The headers if found, None otherwise.
165 | """
166 | # Try to get headers from auth_manager
167 | if hasattr(auth_manager, 'get_headers'):
168 | return auth_manager.get_headers()
169 |
170 | # If auth_manager doesn't have get_headers, try server_config
171 | if hasattr(server_config, 'get_headers'):
172 | return server_config.get_headers()
173 |
174 | # If neither has get_headers, check if auth_manager is actually a ServerConfig
175 | # and server_config is actually an AuthManager (parameters swapped)
176 | if hasattr(server_config, 'get_headers') and not hasattr(auth_manager, 'get_headers'):
177 | return server_config.get_headers()
178 |
179 | logger.error("Cannot find get_headers method in either auth_manager or server_config")
180 | return None
181 |
182 | def create_story(
183 | auth_manager: AuthManager,
184 | server_config: ServerConfig,
185 | params: Dict[str, Any],
186 | ) -> Dict[str, Any]:
187 | """
188 | Create a new story in ServiceNow.
189 |
190 | Args:
191 | auth_manager: The authentication manager.
192 | server_config: The server configuration.
193 | params: The parameters for creating the story.
194 |
195 | Returns:
196 | The created story.
197 | """
198 |
199 | # Unwrap and validate parameters
200 | result = _unwrap_and_validate_params(
201 | params,
202 | CreateStoryParams,
203 | required_fields=["short_description", "acceptance_criteria"]
204 | )
205 |
206 | if not result["success"]:
207 | return result
208 |
209 | validated_params = result["params"]
210 |
211 | # Prepare the request data
212 | data = {
213 | "short_description": validated_params.short_description,
214 | "acceptance_criteria": validated_params.acceptance_criteria,
215 | }
216 |
217 | # Add optional fields if provided
218 | if validated_params.description:
219 | data["description"] = validated_params.description
220 | if validated_params.state:
221 | data["state"] = validated_params.state
222 | if validated_params.assignment_group:
223 | data["assignment_group"] = validated_params.assignment_group
224 | if validated_params.story_points:
225 | data["story_points"] = validated_params.story_points
226 | if validated_params.assigned_to:
227 | data["assigned_to"] = validated_params.assigned_to
228 | if validated_params.epic:
229 | data["epic"] = validated_params.epic
230 | if validated_params.project:
231 | data["project"] = validated_params.project
232 | if validated_params.work_notes:
233 | data["work_notes"] = validated_params.work_notes
234 |
235 | # Get the instance URL
236 | instance_url = _get_instance_url(auth_manager, server_config)
237 | if not instance_url:
238 | return {
239 | "success": False,
240 | "message": "Cannot find instance_url in either server_config or auth_manager",
241 | }
242 |
243 | # Get the headers
244 | headers = _get_headers(auth_manager, server_config)
245 | if not headers:
246 | return {
247 | "success": False,
248 | "message": "Cannot find get_headers method in either auth_manager or server_config",
249 | }
250 |
251 | # Add Content-Type header
252 | headers["Content-Type"] = "application/json"
253 |
254 | # Make the API request
255 | url = f"{instance_url}/api/now/table/rm_story"
256 |
257 | try:
258 | response = requests.post(url, json=data, headers=headers)
259 | response.raise_for_status()
260 |
261 | result = response.json()
262 |
263 | return {
264 | "success": True,
265 | "message": "Story created successfully",
266 | "story": result["result"],
267 | }
268 | except requests.exceptions.RequestException as e:
269 | logger.error(f"Error creating story: {e}")
270 | return {
271 | "success": False,
272 | "message": f"Error creating story: {str(e)}",
273 | }
274 |
275 | def update_story(
276 | auth_manager: AuthManager,
277 | server_config: ServerConfig,
278 | params: Dict[str, Any],
279 | ) -> Dict[str, Any]:
280 | """
281 | Update an existing story in ServiceNow.
282 |
283 | Args:
284 | auth_manager: The authentication manager.
285 | server_config: The server configuration.
286 | params: The parameters for updating the story.
287 |
288 | Returns:
289 | The updated story.
290 | """
291 | # Unwrap and validate parameters
292 | result = _unwrap_and_validate_params(
293 | params,
294 | UpdateStoryParams,
295 | required_fields=["story_id"]
296 | )
297 |
298 | if not result["success"]:
299 | return result
300 |
301 | validated_params = result["params"]
302 |
303 | # Prepare the request data
304 | data = {}
305 |
306 | # Add optional fields if provided
307 | if validated_params.short_description:
308 | data["short_description"] = validated_params.short_description
309 | if validated_params.acceptance_criteria:
310 | data["acceptance_criteria"] = validated_params.acceptance_criteria
311 | if validated_params.description:
312 | data["description"] = validated_params.description
313 | if validated_params.state:
314 | data["state"] = validated_params.state
315 | if validated_params.assignment_group:
316 | data["assignment_group"] = validated_params.assignment_group
317 | if validated_params.story_points:
318 | data["story_points"] = validated_params.story_points
319 | if validated_params.epic:
320 | data["epic"] = validated_params.epic
321 | if validated_params.project:
322 | data["project"] = validated_params.project
323 | if validated_params.assigned_to:
324 | data["assigned_to"] = validated_params.assigned_to
325 | if validated_params.work_notes:
326 | data["work_notes"] = validated_params.work_notes
327 |
328 | # Get the instance URL
329 | instance_url = _get_instance_url(auth_manager, server_config)
330 | if not instance_url:
331 | return {
332 | "success": False,
333 | "message": "Cannot find instance_url in either server_config or auth_manager",
334 | }
335 |
336 | # Get the headers
337 | headers = _get_headers(auth_manager, server_config)
338 | if not headers:
339 | return {
340 | "success": False,
341 | "message": "Cannot find get_headers method in either auth_manager or server_config",
342 | }
343 |
344 | # Add Content-Type header
345 | headers["Content-Type"] = "application/json"
346 |
347 | # Make the API request
348 | url = f"{instance_url}/api/now/table/rm_story/{validated_params.story_id}"
349 |
350 | try:
351 | response = requests.put(url, json=data, headers=headers)
352 | response.raise_for_status()
353 |
354 | result = response.json()
355 |
356 | return {
357 | "success": True,
358 | "message": "Story updated successfully",
359 | "story": result["result"],
360 | }
361 | except requests.exceptions.RequestException as e:
362 | logger.error(f"Error updating story: {e}")
363 | return {
364 | "success": False,
365 | "message": f"Error updating story: {str(e)}",
366 | }
367 |
368 | def list_stories(
369 | auth_manager: AuthManager,
370 | server_config: ServerConfig,
371 | params: Dict[str, Any],
372 | ) -> Dict[str, Any]:
373 | """
374 | List stories from ServiceNow.
375 |
376 | Args:
377 | auth_manager: The authentication manager.
378 | server_config: The server configuration.
379 | params: The parameters for listing stories.
380 |
381 | Returns:
382 | A list of stories.
383 | """
384 | # Unwrap and validate parameters
385 | result = _unwrap_and_validate_params(
386 | params,
387 | ListStoriesParams
388 | )
389 |
390 | if not result["success"]:
391 | return result
392 |
393 | validated_params = result["params"]
394 |
395 | # Build the query
396 | query_parts = []
397 |
398 | if validated_params.state:
399 | query_parts.append(f"state={validated_params.state}")
400 | if validated_params.assignment_group:
401 | query_parts.append(f"assignment_group={validated_params.assignment_group}")
402 |
403 | # Handle timeframe filtering
404 | if validated_params.timeframe:
405 | now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
406 | if validated_params.timeframe == "upcoming":
407 | query_parts.append(f"start_date>{now}")
408 | elif validated_params.timeframe == "in-progress":
409 | query_parts.append(f"start_date<{now}^end_date>{now}")
410 | elif validated_params.timeframe == "completed":
411 | query_parts.append(f"end_date<{now}")
412 |
413 | # Add any additional query string
414 | if validated_params.query:
415 | query_parts.append(validated_params.query)
416 |
417 | # Combine query parts
418 | query = "^".join(query_parts) if query_parts else ""
419 |
420 | # Get the instance URL
421 | instance_url = _get_instance_url(auth_manager, server_config)
422 | if not instance_url:
423 | return {
424 | "success": False,
425 | "message": "Cannot find instance_url in either server_config or auth_manager",
426 | }
427 |
428 | # Get the headers
429 | headers = _get_headers(auth_manager, server_config)
430 | if not headers:
431 | return {
432 | "success": False,
433 | "message": "Cannot find get_headers method in either auth_manager or server_config",
434 | }
435 |
436 | # Make the API request
437 | url = f"{instance_url}/api/now/table/rm_story"
438 |
439 | params = {
440 | "sysparm_limit": validated_params.limit,
441 | "sysparm_offset": validated_params.offset,
442 | "sysparm_query": query,
443 | "sysparm_display_value": "true",
444 | }
445 |
446 | try:
447 | response = requests.get(url, headers=headers, params=params)
448 | response.raise_for_status()
449 |
450 | result = response.json()
451 |
452 | # Handle the case where result["result"] is a list
453 | stories = result.get("result", [])
454 | count = len(stories)
455 |
456 | return {
457 | "success": True,
458 | "stories": stories,
459 | "count": count,
460 | "total": count, # Use count as total if total is not provided
461 | }
462 | except requests.exceptions.RequestException as e:
463 | logger.error(f"Error listing stories: {e}")
464 | return {
465 | "success": False,
466 | "message": f"Error listing stories: {str(e)}",
467 | }
468 |
469 | def list_story_dependencies(
470 | auth_manager: AuthManager,
471 | server_config: ServerConfig,
472 | params: Dict[str, Any],
473 | ) -> Dict[str, Any]:
474 | """
475 | List story dependencies from ServiceNow.
476 |
477 | Args:
478 | auth_manager: The authentication manager.
479 | server_config: The server configuration.
480 | params: The parameters for listing story dependencies.
481 |
482 | Returns:
483 | A list of story dependencies.
484 | """
485 | # Unwrap and validate parameters
486 | result = _unwrap_and_validate_params(
487 | params,
488 | ListStoryDependenciesParams
489 | )
490 |
491 | if not result["success"]:
492 | return result
493 |
494 | validated_params = result["params"]
495 |
496 | # Build the query
497 | query_parts = []
498 |
499 | if validated_params.dependent_story:
500 | query_parts.append(f"dependent_story={validated_params.dependent_story}")
501 | if validated_params.prerequisite_story:
502 | query_parts.append(f"prerequisite_story={validated_params.prerequisite_story}")
503 |
504 | # Add any additional query string
505 | if validated_params.query:
506 | query_parts.append(validated_params.query)
507 |
508 | # Combine query parts
509 | query = "^".join(query_parts) if query_parts else ""
510 |
511 | # Get the instance URL
512 | instance_url = _get_instance_url(auth_manager, server_config)
513 | if not instance_url:
514 | return {
515 | "success": False,
516 | "message": "Cannot find instance_url in either server_config or auth_manager",
517 | }
518 |
519 | # Get the headers
520 | headers = _get_headers(auth_manager, server_config)
521 | if not headers:
522 | return {
523 | "success": False,
524 | "message": "Cannot find get_headers method in either auth_manager or server_config",
525 | }
526 |
527 | # Make the API request
528 | url = f"{instance_url}/api/now/table/m2m_story_dependencies"
529 |
530 | params = {
531 | "sysparm_limit": validated_params.limit,
532 | "sysparm_offset": validated_params.offset,
533 | "sysparm_query": query,
534 | "sysparm_display_value": "true",
535 | }
536 |
537 | try:
538 | response = requests.get(url, headers=headers, params=params)
539 | response.raise_for_status()
540 |
541 | result = response.json()
542 |
543 | # Handle the case where result["result"] is a list
544 | story_dependencies = result.get("result", [])
545 | count = len(story_dependencies)
546 |
547 | return {
548 | "success": True,
549 | "story_dependencies": story_dependencies,
550 | "count": count,
551 | "total": count, # Use count as total if total is not provided
552 | }
553 | except requests.exceptions.RequestException as e:
554 | logger.error(f"Error listing story dependencies: {e}")
555 | return {
556 | "success": False,
557 | "message": f"Error listing story dependencies: {str(e)}",
558 | }
559 |
560 | def create_story_dependency(
561 | auth_manager: AuthManager,
562 | server_config: ServerConfig,
563 | params: Dict[str, Any],
564 | ) -> Dict[str, Any]:
565 | """
566 | Create a dependency between two stories in ServiceNow.
567 |
568 | Args:
569 | auth_manager: The authentication manager.
570 | server_config: The server configuration.
571 | params: The parameters for creating a story dependency.
572 |
573 | Returns:
574 | The created story dependency.
575 | """
576 | # Unwrap and validate parameters
577 | result = _unwrap_and_validate_params(
578 | params,
579 | CreateStoryDependencyParams,
580 | required_fields=["dependent_story", "prerequisite_story"]
581 | )
582 |
583 | if not result["success"]:
584 | return result
585 |
586 | validated_params = result["params"]
587 |
588 | # Prepare the request data
589 | data = {
590 | "dependent_story": validated_params.dependent_story,
591 | "prerequisite_story": validated_params.prerequisite_story,
592 | }
593 |
594 | # Get the instance URL
595 | instance_url = _get_instance_url(auth_manager, server_config)
596 | if not instance_url:
597 | return {
598 | "success": False,
599 | "message": "Cannot find instance_url in either server_config or auth_manager",
600 | }
601 |
602 | # Get the headers
603 | headers = _get_headers(auth_manager, server_config)
604 | if not headers:
605 | return {
606 | "success": False,
607 | "message": "Cannot find get_headers method in either auth_manager or server_config",
608 | }
609 |
610 | # Add Content-Type header
611 | headers["Content-Type"] = "application/json"
612 |
613 | # Make the API request
614 | url = f"{instance_url}/api/now/table/m2m_story_dependencies"
615 |
616 | try:
617 | response = requests.post(url, json=data, headers=headers)
618 | response.raise_for_status()
619 |
620 | result = response.json()
621 | return {
622 | "success": True,
623 | "message": "Story dependency created successfully",
624 | "story_dependency": result["result"],
625 | }
626 | except requests.exceptions.RequestException as e:
627 | logger.error(f"Error creating story dependency: {e}")
628 | return {
629 | "success": False,
630 | "message": f"Error creating story dependency: {str(e)}",
631 | }
632 | def delete_story_dependency(
633 | auth_manager: AuthManager,
634 | server_config: ServerConfig,
635 | params: Dict[str, Any],
636 | ) -> Dict[str, Any]:
637 | """
638 | Delete a story dependency in ServiceNow.
639 |
640 | Args:
641 | auth_manager: The authentication manager.
642 | server_config: The server configuration.
643 | params: The parameters for deleting a story dependency.
644 |
645 | Returns:
646 | The deleted story dependency.
647 | """
648 | # Unwrap and validate parameters
649 | result = _unwrap_and_validate_params(
650 | params,
651 | DeleteStoryDependencyParams,
652 | required_fields=["dependency_id"]
653 | )
654 |
655 | if not result["success"]:
656 | return result
657 |
658 | validated_params = result["params"]
659 |
660 | # Get the instance URL
661 | instance_url = _get_instance_url(auth_manager, server_config)
662 | if not instance_url:
663 | return {
664 | "success": False,
665 | "message": "Cannot find instance_url in either server_config or auth_manager",
666 | }
667 |
668 | # Get the headers
669 | headers = _get_headers(auth_manager, server_config)
670 | if not headers:
671 | return {
672 | "success": False,
673 | "message": "Cannot find get_headers method in either auth_manager or server_config",
674 | }
675 |
676 | # Make the API request
677 | url = f"{instance_url}/api/now/table/m2m_story_dependencies/{validated_params.dependency_id}"
678 |
679 | try:
680 | response = requests.delete(url, headers=headers)
681 | response.raise_for_status()
682 |
683 | return {
684 | "success": True,
685 | "message": "Story dependency deleted successfully",
686 | }
687 | except requests.exceptions.RequestException as e:
688 | logger.error(f"Error deleting story dependency: {e}")
689 | return {
690 | "success": False,
691 | "message": f"Error deleting story dependency: {str(e)}",
692 | }
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/changeset_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Changeset tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing changesets in ServiceNow.
5 | """
6 |
7 | import logging
8 | from typing import Any, Dict, List, Optional, Type, TypeVar, Union
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 | # Type variable for Pydantic models
19 | T = TypeVar('T', bound=BaseModel)
20 |
21 |
22 | class ListChangesetsParams(BaseModel):
23 | """Parameters for listing changesets."""
24 |
25 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
26 | offset: Optional[int] = Field(0, description="Offset to start from")
27 | state: Optional[str] = Field(None, description="Filter by state")
28 | application: Optional[str] = Field(None, description="Filter by application")
29 | developer: Optional[str] = Field(None, description="Filter by developer")
30 | timeframe: Optional[str] = Field(None, description="Filter by timeframe (recent, last_week, last_month)")
31 | query: Optional[str] = Field(None, description="Additional query string")
32 |
33 |
34 | class GetChangesetDetailsParams(BaseModel):
35 | """Parameters for getting changeset details."""
36 |
37 | changeset_id: str = Field(..., description="Changeset ID or sys_id")
38 |
39 |
40 | class CreateChangesetParams(BaseModel):
41 | """Parameters for creating a changeset."""
42 |
43 | name: str = Field(..., description="Name of the changeset")
44 | description: Optional[str] = Field(None, description="Description of the changeset")
45 | application: str = Field(..., description="Application the changeset belongs to")
46 | developer: Optional[str] = Field(None, description="Developer responsible for the changeset")
47 |
48 |
49 | class UpdateChangesetParams(BaseModel):
50 | """Parameters for updating a changeset."""
51 |
52 | changeset_id: str = Field(..., description="Changeset ID or sys_id")
53 | name: Optional[str] = Field(None, description="Name of the changeset")
54 | description: Optional[str] = Field(None, description="Description of the changeset")
55 | state: Optional[str] = Field(None, description="State of the changeset")
56 | developer: Optional[str] = Field(None, description="Developer responsible for the changeset")
57 |
58 |
59 | class CommitChangesetParams(BaseModel):
60 | """Parameters for committing a changeset."""
61 |
62 | changeset_id: str = Field(..., description="Changeset ID or sys_id")
63 | commit_message: Optional[str] = Field(None, description="Commit message")
64 |
65 |
66 | class PublishChangesetParams(BaseModel):
67 | """Parameters for publishing a changeset."""
68 |
69 | changeset_id: str = Field(..., description="Changeset ID or sys_id")
70 | publish_notes: Optional[str] = Field(None, description="Notes for publishing")
71 |
72 |
73 | class AddFileToChangesetParams(BaseModel):
74 | """Parameters for adding a file to a changeset."""
75 |
76 | changeset_id: str = Field(..., description="Changeset ID or sys_id")
77 | file_path: str = Field(..., description="Path of the file to add")
78 | file_content: str = Field(..., description="Content of the file")
79 |
80 |
81 | def _unwrap_and_validate_params(
82 | params: Union[Dict[str, Any], BaseModel],
83 | model_class: Type[T],
84 | required_fields: Optional[List[str]] = None
85 | ) -> Dict[str, Any]:
86 | """
87 | Unwrap and validate parameters.
88 |
89 | Args:
90 | params: The parameters to unwrap and validate. Can be a dictionary or a Pydantic model.
91 | model_class: The Pydantic model class to validate against.
92 | required_fields: List of fields that must be present.
93 |
94 | Returns:
95 | A dictionary with success status and validated parameters or error message.
96 | """
97 | try:
98 | # Handle case where params is already a Pydantic model
99 | if isinstance(params, BaseModel):
100 | # If it's already the correct model class, use it directly
101 | if isinstance(params, model_class):
102 | model_instance = params
103 | # Otherwise, convert to dict and create new instance
104 | else:
105 | model_instance = model_class(**params.dict())
106 | # Handle dictionary case
107 | else:
108 | # Create model instance
109 | model_instance = model_class(**params)
110 |
111 | # Check required fields
112 | if required_fields:
113 | missing_fields = []
114 | for field in required_fields:
115 | if getattr(model_instance, field, None) is None:
116 | missing_fields.append(field)
117 |
118 | if missing_fields:
119 | return {
120 | "success": False,
121 | "message": f"Missing required fields: {', '.join(missing_fields)}",
122 | }
123 |
124 | return {
125 | "success": True,
126 | "params": model_instance,
127 | }
128 | except Exception as e:
129 | return {
130 | "success": False,
131 | "message": f"Invalid parameters: {str(e)}",
132 | }
133 |
134 |
135 | def _get_instance_url(auth_manager: AuthManager, server_config: ServerConfig) -> Optional[str]:
136 | """
137 | Get the instance URL from either auth_manager or server_config.
138 |
139 | Args:
140 | auth_manager: The authentication manager.
141 | server_config: The server configuration.
142 |
143 | Returns:
144 | The instance URL or None if not found.
145 | """
146 | # Try to get instance_url from server_config
147 | if hasattr(server_config, 'instance_url'):
148 | return server_config.instance_url
149 |
150 | # Try to get instance_url from auth_manager
151 | if hasattr(auth_manager, 'instance_url'):
152 | return auth_manager.instance_url
153 |
154 | # If neither has instance_url, check if auth_manager is actually a ServerConfig
155 | # and server_config is actually an AuthManager (parameters swapped)
156 | if hasattr(server_config, 'get_headers') and not hasattr(auth_manager, 'get_headers'):
157 | if hasattr(auth_manager, 'instance_url'):
158 | return auth_manager.instance_url
159 |
160 | logger.error("Cannot find instance_url in either auth_manager or server_config")
161 | return None
162 |
163 |
164 | def _get_headers(auth_manager: AuthManager, server_config: ServerConfig) -> Optional[Dict[str, str]]:
165 | """
166 | Get the headers from either auth_manager or server_config.
167 |
168 | Args:
169 | auth_manager: The authentication manager.
170 | server_config: The server configuration.
171 |
172 | Returns:
173 | The headers or None if not found.
174 | """
175 | # Try to get headers from auth_manager
176 | if hasattr(auth_manager, 'get_headers'):
177 | return auth_manager.get_headers()
178 |
179 | # Try to get headers from server_config
180 | if hasattr(server_config, 'get_headers'):
181 | return server_config.get_headers()
182 |
183 | # If neither has get_headers, check if auth_manager is actually a ServerConfig
184 | # and server_config is actually an AuthManager (parameters swapped)
185 | if hasattr(server_config, 'get_headers') and not hasattr(auth_manager, 'get_headers'):
186 | return server_config.get_headers()
187 |
188 | logger.error("Cannot find get_headers method in either auth_manager or server_config")
189 | return None
190 |
191 |
192 | def list_changesets(
193 | auth_manager: AuthManager,
194 | server_config: ServerConfig,
195 | params: Union[Dict[str, Any], ListChangesetsParams],
196 | ) -> Dict[str, Any]:
197 | """
198 | List changesets from ServiceNow.
199 |
200 | Args:
201 | auth_manager: The authentication manager.
202 | server_config: The server configuration.
203 | params: The parameters for listing changesets. Can be a dictionary or a ListChangesetsParams object.
204 |
205 | Returns:
206 | A list of changesets.
207 | """
208 | # Unwrap and validate parameters
209 | result = _unwrap_and_validate_params(params, ListChangesetsParams)
210 |
211 | if not result["success"]:
212 | return result
213 |
214 | validated_params = result["params"]
215 |
216 | # Get the instance URL
217 | instance_url = _get_instance_url(auth_manager, server_config)
218 | if not instance_url:
219 | return {
220 | "success": False,
221 | "message": "Cannot find instance_url in either server_config or auth_manager",
222 | }
223 |
224 | # Get the headers
225 | headers = _get_headers(auth_manager, server_config)
226 | if not headers:
227 | return {
228 | "success": False,
229 | "message": "Cannot find get_headers method in either auth_manager or server_config",
230 | }
231 |
232 | # Build query parameters
233 | query_params = {
234 | "sysparm_limit": validated_params.limit,
235 | "sysparm_offset": validated_params.offset,
236 | }
237 |
238 | # Build sysparm_query
239 | query_parts = []
240 |
241 | if validated_params.state:
242 | query_parts.append(f"state={validated_params.state}")
243 |
244 | if validated_params.application:
245 | query_parts.append(f"application={validated_params.application}")
246 |
247 | if validated_params.developer:
248 | query_parts.append(f"developer={validated_params.developer}")
249 |
250 | if validated_params.timeframe:
251 | if validated_params.timeframe == "recent":
252 | query_parts.append("sys_created_onONLast 7 days@javascript:gs.beginningOfLast7Days()@javascript:gs.endOfToday()")
253 | elif validated_params.timeframe == "last_week":
254 | query_parts.append("sys_created_onONLast week@javascript:gs.beginningOfLastWeek()@javascript:gs.endOfLastWeek()")
255 | elif validated_params.timeframe == "last_month":
256 | query_parts.append("sys_created_onONLast month@javascript:gs.beginningOfLastMonth()@javascript:gs.endOfLastMonth()")
257 |
258 | if validated_params.query:
259 | query_parts.append(validated_params.query)
260 |
261 | if query_parts:
262 | query_params["sysparm_query"] = "^".join(query_parts)
263 |
264 | # Make the API request
265 | url = f"{instance_url}/api/now/table/sys_update_set"
266 |
267 | try:
268 | response = requests.get(url, params=query_params, headers=headers)
269 | response.raise_for_status()
270 |
271 | result = response.json()
272 |
273 | return {
274 | "success": True,
275 | "changesets": result.get("result", []),
276 | "count": len(result.get("result", [])),
277 | }
278 | except requests.exceptions.RequestException as e:
279 | logger.error(f"Error listing changesets: {e}")
280 | return {
281 | "success": False,
282 | "message": f"Error listing changesets: {str(e)}",
283 | }
284 |
285 |
286 | def get_changeset_details(
287 | auth_manager: AuthManager,
288 | server_config: ServerConfig,
289 | params: Union[Dict[str, Any], GetChangesetDetailsParams],
290 | ) -> Dict[str, Any]:
291 | """
292 | Get detailed information about a specific changeset.
293 |
294 | Args:
295 | auth_manager: The authentication manager.
296 | server_config: The server configuration.
297 | params: The parameters for getting changeset details. Can be a dictionary or a GetChangesetDetailsParams object.
298 |
299 | Returns:
300 | Detailed information about the changeset.
301 | """
302 | # Unwrap and validate parameters
303 | result = _unwrap_and_validate_params(
304 | params,
305 | GetChangesetDetailsParams,
306 | required_fields=["changeset_id"]
307 | )
308 |
309 | if not result["success"]:
310 | return result
311 |
312 | validated_params = result["params"]
313 |
314 | # Get the instance URL
315 | instance_url = _get_instance_url(auth_manager, server_config)
316 | if not instance_url:
317 | return {
318 | "success": False,
319 | "message": "Cannot find instance_url in either server_config or auth_manager",
320 | }
321 |
322 | # Get the headers
323 | headers = _get_headers(auth_manager, server_config)
324 | if not headers:
325 | return {
326 | "success": False,
327 | "message": "Cannot find get_headers method in either auth_manager or server_config",
328 | }
329 |
330 | # Make the API request
331 | url = f"{instance_url}/api/now/table/sys_update_set/{validated_params.changeset_id}"
332 |
333 | try:
334 | response = requests.get(url, headers=headers)
335 | response.raise_for_status()
336 |
337 | result = response.json()
338 |
339 | # Get the changeset details
340 | changeset = result.get("result", {})
341 |
342 | # Get the changes in this changeset
343 | changes_url = f"{instance_url}/api/now/table/sys_update_xml"
344 | changes_params = {
345 | "sysparm_query": f"update_set={validated_params.changeset_id}",
346 | }
347 |
348 | changes_response = requests.get(changes_url, params=changes_params, headers=headers)
349 | changes_response.raise_for_status()
350 |
351 | changes_result = changes_response.json()
352 | changes = changes_result.get("result", [])
353 |
354 | return {
355 | "success": True,
356 | "changeset": changeset,
357 | "changes": changes,
358 | "change_count": len(changes),
359 | }
360 | except requests.exceptions.RequestException as e:
361 | logger.error(f"Error getting changeset details: {e}")
362 | return {
363 | "success": False,
364 | "message": f"Error getting changeset details: {str(e)}",
365 | }
366 |
367 |
368 | def create_changeset(
369 | auth_manager: AuthManager,
370 | server_config: ServerConfig,
371 | params: Union[Dict[str, Any], CreateChangesetParams],
372 | ) -> Dict[str, Any]:
373 | """
374 | Create a new changeset in ServiceNow.
375 |
376 | Args:
377 | auth_manager: The authentication manager.
378 | server_config: The server configuration.
379 | params: The parameters for creating a changeset. Can be a dictionary or a CreateChangesetParams object.
380 |
381 | Returns:
382 | The created changeset.
383 | """
384 | # Unwrap and validate parameters
385 | result = _unwrap_and_validate_params(
386 | params,
387 | CreateChangesetParams,
388 | required_fields=["name", "application"]
389 | )
390 |
391 | if not result["success"]:
392 | return result
393 |
394 | validated_params = result["params"]
395 |
396 | # Prepare the request data
397 | data = {
398 | "name": validated_params.name,
399 | "application": validated_params.application,
400 | }
401 |
402 | # Add optional fields if provided
403 | if validated_params.description:
404 | data["description"] = validated_params.description
405 | if validated_params.developer:
406 | data["developer"] = validated_params.developer
407 |
408 | # Get the instance URL
409 | instance_url = _get_instance_url(auth_manager, server_config)
410 | if not instance_url:
411 | return {
412 | "success": False,
413 | "message": "Cannot find instance_url in either server_config or auth_manager",
414 | }
415 |
416 | # Get the headers
417 | headers = _get_headers(auth_manager, server_config)
418 | if not headers:
419 | return {
420 | "success": False,
421 | "message": "Cannot find get_headers method in either auth_manager or server_config",
422 | }
423 |
424 | # Add Content-Type header
425 | headers["Content-Type"] = "application/json"
426 |
427 | # Make the API request
428 | url = f"{instance_url}/api/now/table/sys_update_set"
429 |
430 | try:
431 | response = requests.post(url, json=data, headers=headers)
432 | response.raise_for_status()
433 |
434 | result = response.json()
435 |
436 | return {
437 | "success": True,
438 | "message": "Changeset created successfully",
439 | "changeset": result["result"],
440 | }
441 | except requests.exceptions.RequestException as e:
442 | logger.error(f"Error creating changeset: {e}")
443 | return {
444 | "success": False,
445 | "message": f"Error creating changeset: {str(e)}",
446 | }
447 |
448 |
449 | def update_changeset(
450 | auth_manager: AuthManager,
451 | server_config: ServerConfig,
452 | params: Union[Dict[str, Any], UpdateChangesetParams],
453 | ) -> Dict[str, Any]:
454 | """
455 | Update an existing changeset in ServiceNow.
456 |
457 | Args:
458 | auth_manager: The authentication manager.
459 | server_config: The server configuration.
460 | params: The parameters for updating a changeset. Can be a dictionary or a UpdateChangesetParams object.
461 |
462 | Returns:
463 | The updated changeset.
464 | """
465 | # Unwrap and validate parameters
466 | result = _unwrap_and_validate_params(
467 | params,
468 | UpdateChangesetParams,
469 | required_fields=["changeset_id"]
470 | )
471 |
472 | if not result["success"]:
473 | return result
474 |
475 | validated_params = result["params"]
476 |
477 | # Prepare the request data
478 | data = {}
479 |
480 | # Add optional fields if provided
481 | if validated_params.name:
482 | data["name"] = validated_params.name
483 | if validated_params.description:
484 | data["description"] = validated_params.description
485 | if validated_params.state:
486 | data["state"] = validated_params.state
487 | if validated_params.developer:
488 | data["developer"] = validated_params.developer
489 |
490 | # If no fields to update, return error
491 | if not data:
492 | return {
493 | "success": False,
494 | "message": "No fields to update",
495 | }
496 |
497 | # Get the instance URL
498 | instance_url = _get_instance_url(auth_manager, server_config)
499 | if not instance_url:
500 | return {
501 | "success": False,
502 | "message": "Cannot find instance_url in either server_config or auth_manager",
503 | }
504 |
505 | # Get the headers
506 | headers = _get_headers(auth_manager, server_config)
507 | if not headers:
508 | return {
509 | "success": False,
510 | "message": "Cannot find get_headers method in either auth_manager or server_config",
511 | }
512 |
513 | # Add Content-Type header
514 | headers["Content-Type"] = "application/json"
515 |
516 | # Make the API request
517 | url = f"{instance_url}/api/now/table/sys_update_set/{validated_params.changeset_id}"
518 |
519 | try:
520 | response = requests.patch(url, json=data, headers=headers)
521 | response.raise_for_status()
522 |
523 | result = response.json()
524 |
525 | return {
526 | "success": True,
527 | "message": "Changeset updated successfully",
528 | "changeset": result["result"],
529 | }
530 | except requests.exceptions.RequestException as e:
531 | logger.error(f"Error updating changeset: {e}")
532 | return {
533 | "success": False,
534 | "message": f"Error updating changeset: {str(e)}",
535 | }
536 |
537 |
538 | def commit_changeset(
539 | auth_manager: AuthManager,
540 | server_config: ServerConfig,
541 | params: Union[Dict[str, Any], CommitChangesetParams],
542 | ) -> Dict[str, Any]:
543 | """
544 | Commit a changeset in ServiceNow.
545 |
546 | Args:
547 | auth_manager: The authentication manager.
548 | server_config: The server configuration.
549 | params: The parameters for committing a changeset. Can be a dictionary or a CommitChangesetParams object.
550 |
551 | Returns:
552 | The committed changeset.
553 | """
554 | # Unwrap and validate parameters
555 | result = _unwrap_and_validate_params(
556 | params,
557 | CommitChangesetParams,
558 | required_fields=["changeset_id"]
559 | )
560 |
561 | if not result["success"]:
562 | return result
563 |
564 | validated_params = result["params"]
565 |
566 | # Prepare the request data
567 | data = {
568 | "state": "complete",
569 | }
570 |
571 | # Add commit message if provided
572 | if validated_params.commit_message:
573 | data["description"] = validated_params.commit_message
574 |
575 | # Get the instance URL
576 | instance_url = _get_instance_url(auth_manager, server_config)
577 | if not instance_url:
578 | return {
579 | "success": False,
580 | "message": "Cannot find instance_url in either server_config or auth_manager",
581 | }
582 |
583 | # Get the headers
584 | headers = _get_headers(auth_manager, server_config)
585 | if not headers:
586 | return {
587 | "success": False,
588 | "message": "Cannot find get_headers method in either auth_manager or server_config",
589 | }
590 |
591 | # Add Content-Type header
592 | headers["Content-Type"] = "application/json"
593 |
594 | # Make the API request
595 | url = f"{instance_url}/api/now/table/sys_update_set/{validated_params.changeset_id}"
596 |
597 | try:
598 | response = requests.patch(url, json=data, headers=headers)
599 | response.raise_for_status()
600 |
601 | result = response.json()
602 |
603 | return {
604 | "success": True,
605 | "message": "Changeset committed successfully",
606 | "changeset": result["result"],
607 | }
608 | except requests.exceptions.RequestException as e:
609 | logger.error(f"Error committing changeset: {e}")
610 | return {
611 | "success": False,
612 | "message": f"Error committing changeset: {str(e)}",
613 | }
614 |
615 |
616 | def publish_changeset(
617 | auth_manager: AuthManager,
618 | server_config: ServerConfig,
619 | params: Union[Dict[str, Any], PublishChangesetParams],
620 | ) -> Dict[str, Any]:
621 | """
622 | Publish a changeset in ServiceNow.
623 |
624 | Args:
625 | auth_manager: The authentication manager.
626 | server_config: The server configuration.
627 | params: The parameters for publishing a changeset. Can be a dictionary or a PublishChangesetParams object.
628 |
629 | Returns:
630 | The published changeset.
631 | """
632 | # Unwrap and validate parameters
633 | result = _unwrap_and_validate_params(
634 | params,
635 | PublishChangesetParams,
636 | required_fields=["changeset_id"]
637 | )
638 |
639 | if not result["success"]:
640 | return result
641 |
642 | validated_params = result["params"]
643 |
644 | # Get the instance URL
645 | instance_url = _get_instance_url(auth_manager, server_config)
646 | if not instance_url:
647 | return {
648 | "success": False,
649 | "message": "Cannot find instance_url in either server_config or auth_manager",
650 | }
651 |
652 | # Get the headers
653 | headers = _get_headers(auth_manager, server_config)
654 | if not headers:
655 | return {
656 | "success": False,
657 | "message": "Cannot find get_headers method in either auth_manager or server_config",
658 | }
659 |
660 | # Add Content-Type header
661 | headers["Content-Type"] = "application/json"
662 |
663 | # Prepare the request data for the publish action
664 | data = {
665 | "state": "published",
666 | }
667 |
668 | # Add publish notes if provided
669 | if validated_params.publish_notes:
670 | data["description"] = validated_params.publish_notes
671 |
672 | # Make the API request
673 | url = f"{instance_url}/api/now/table/sys_update_set/{validated_params.changeset_id}"
674 |
675 | try:
676 | response = requests.patch(url, json=data, headers=headers)
677 | response.raise_for_status()
678 |
679 | result = response.json()
680 |
681 | return {
682 | "success": True,
683 | "message": "Changeset published successfully",
684 | "changeset": result["result"],
685 | }
686 | except requests.exceptions.RequestException as e:
687 | logger.error(f"Error publishing changeset: {e}")
688 | return {
689 | "success": False,
690 | "message": f"Error publishing changeset: {str(e)}",
691 | }
692 |
693 |
694 | def add_file_to_changeset(
695 | auth_manager: AuthManager,
696 | server_config: ServerConfig,
697 | params: Union[Dict[str, Any], AddFileToChangesetParams],
698 | ) -> Dict[str, Any]:
699 | """
700 | Add a file to a changeset in ServiceNow.
701 |
702 | Args:
703 | auth_manager: The authentication manager.
704 | server_config: The server configuration.
705 | params: The parameters for adding a file to a changeset. Can be a dictionary or a AddFileToChangesetParams object.
706 |
707 | Returns:
708 | The result of the add file operation.
709 | """
710 | # Unwrap and validate parameters
711 | result = _unwrap_and_validate_params(
712 | params,
713 | AddFileToChangesetParams,
714 | required_fields=["changeset_id", "file_path", "file_content"]
715 | )
716 |
717 | if not result["success"]:
718 | return result
719 |
720 | validated_params = result["params"]
721 |
722 | # Get the instance URL
723 | instance_url = _get_instance_url(auth_manager, server_config)
724 | if not instance_url:
725 | return {
726 | "success": False,
727 | "message": "Cannot find instance_url in either server_config or auth_manager",
728 | }
729 |
730 | # Get the headers
731 | headers = _get_headers(auth_manager, server_config)
732 | if not headers:
733 | return {
734 | "success": False,
735 | "message": "Cannot find get_headers method in either auth_manager or server_config",
736 | }
737 |
738 | # Add Content-Type header
739 | headers["Content-Type"] = "application/json"
740 |
741 | # Prepare the request data for adding a file
742 | data = {
743 | "update_set": validated_params.changeset_id,
744 | "name": validated_params.file_path,
745 | "payload": validated_params.file_content,
746 | "type": "file",
747 | }
748 |
749 | # Make the API request
750 | url = f"{instance_url}/api/now/table/sys_update_xml"
751 |
752 | try:
753 | response = requests.post(url, json=data, headers=headers)
754 | response.raise_for_status()
755 |
756 | result = response.json()
757 |
758 | return {
759 | "success": True,
760 | "message": "File added to changeset successfully",
761 | "file": result["result"],
762 | }
763 | except requests.exceptions.RequestException as e:
764 | logger.error(f"Error adding file to changeset: {e}")
765 | return {
766 | "success": False,
767 | "message": f"Error adding file to changeset: {str(e)}",
768 | }
```
--------------------------------------------------------------------------------
/tests/test_knowledge_base.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for the knowledge base tools.
3 |
4 | This module contains tests for the knowledge base tools in the ServiceNow MCP server.
5 | """
6 |
7 | import unittest
8 | from unittest.mock import MagicMock, patch
9 |
10 | import requests
11 |
12 | from servicenow_mcp.auth.auth_manager import AuthManager
13 | from servicenow_mcp.tools.knowledge_base import (
14 | CreateArticleParams,
15 | CreateCategoryParams,
16 | CreateKnowledgeBaseParams,
17 | GetArticleParams,
18 | ListArticlesParams,
19 | ListKnowledgeBasesParams,
20 | PublishArticleParams,
21 | UpdateArticleParams,
22 | ListCategoriesParams,
23 | KnowledgeBaseResponse,
24 | CategoryResponse,
25 | ArticleResponse,
26 | create_article,
27 | create_category,
28 | create_knowledge_base,
29 | get_article,
30 | list_articles,
31 | list_knowledge_bases,
32 | publish_article,
33 | update_article,
34 | list_categories,
35 | )
36 | from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
37 |
38 |
39 | class TestKnowledgeBaseTools(unittest.TestCase):
40 | """Tests for the knowledge base tools."""
41 |
42 | def setUp(self):
43 | """Set up test fixtures."""
44 | auth_config = AuthConfig(
45 | type=AuthType.BASIC,
46 | basic=BasicAuthConfig(
47 | username="test_user",
48 | password="test_password"
49 | )
50 | )
51 | self.server_config = ServerConfig(
52 | instance_url="https://test.service-now.com",
53 | auth=auth_config,
54 | )
55 | self.auth_manager = MagicMock(spec=AuthManager)
56 | self.auth_manager.get_headers.return_value = {
57 | "Authorization": "Bearer test",
58 | "Content-Type": "application/json",
59 | }
60 |
61 | @patch("servicenow_mcp.tools.knowledge_base.requests.post")
62 | def test_create_knowledge_base(self, mock_post):
63 | """Test creating a knowledge base."""
64 | # Mock response
65 | mock_response = MagicMock()
66 | mock_response.json.return_value = {
67 | "result": {
68 | "sys_id": "kb001",
69 | "title": "Test Knowledge Base",
70 | "description": "Test Description",
71 | "owner": "admin",
72 | "kb_managers": "it_managers",
73 | "workflow_publish": "Knowledge - Instant Publish",
74 | "workflow_retire": "Knowledge - Instant Retire",
75 | }
76 | }
77 | mock_response.status_code = 200
78 | mock_post.return_value = mock_response
79 |
80 | # Call the method
81 | params = CreateKnowledgeBaseParams(
82 | title="Test Knowledge Base",
83 | description="Test Description",
84 | owner="admin",
85 | managers="it_managers"
86 | )
87 | result = create_knowledge_base(self.server_config, self.auth_manager, params)
88 |
89 | # Verify the result
90 | self.assertTrue(result.success)
91 | self.assertEqual("kb001", result.kb_id)
92 | self.assertEqual("Test Knowledge Base", result.kb_name)
93 |
94 | # Verify the request
95 | mock_post.assert_called_once()
96 | args, kwargs = mock_post.call_args
97 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge_base", args[0])
98 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
99 | self.assertEqual("Test Knowledge Base", kwargs["json"]["title"])
100 | self.assertEqual("Test Description", kwargs["json"]["description"])
101 | self.assertEqual("admin", kwargs["json"]["owner"])
102 | self.assertEqual("it_managers", kwargs["json"]["kb_managers"])
103 |
104 | @patch("servicenow_mcp.tools.knowledge_base.requests.post")
105 | def test_create_category(self, mock_post):
106 | """Test creating a category."""
107 | # Mock response
108 | mock_response = MagicMock()
109 | mock_response.json.return_value = {
110 | "result": {
111 | "sys_id": "cat001",
112 | "label": "Test Category",
113 | "description": "Test Category Description",
114 | "kb_knowledge_base": "kb001",
115 | "parent": "",
116 | "active": "true",
117 | }
118 | }
119 | mock_response.status_code = 200
120 | mock_post.return_value = mock_response
121 |
122 | # Call the method
123 | params = CreateCategoryParams(
124 | title="Test Category",
125 | description="Test Category Description",
126 | knowledge_base="kb001",
127 | active=True
128 | )
129 | result = create_category(self.server_config, self.auth_manager, params)
130 |
131 | # Verify the result
132 | self.assertTrue(result.success)
133 | self.assertEqual("cat001", result.category_id)
134 | self.assertEqual("Test Category", result.category_name)
135 |
136 | # Verify the request
137 | mock_post.assert_called_once()
138 | args, kwargs = mock_post.call_args
139 | self.assertEqual(f"{self.server_config.api_url}/table/kb_category", args[0])
140 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
141 | self.assertEqual("Test Category", kwargs["json"]["label"])
142 | self.assertEqual("Test Category Description", kwargs["json"]["description"])
143 | self.assertEqual("kb001", kwargs["json"]["kb_knowledge_base"])
144 | self.assertEqual("true", kwargs["json"]["active"])
145 |
146 | @patch("servicenow_mcp.tools.knowledge_base.requests.post")
147 | def test_create_article(self, mock_post):
148 | """Test creating a knowledge article."""
149 | # Mock response
150 | mock_response = MagicMock()
151 | mock_response.json.return_value = {
152 | "result": {
153 | "sys_id": "art001",
154 | "short_description": "Test Article",
155 | "text": "This is a test article content",
156 | "kb_knowledge_base": "kb001",
157 | "kb_category": "cat001",
158 | "article_type": "text",
159 | "keywords": "test,article,knowledge",
160 | "workflow_state": "draft",
161 | }
162 | }
163 | mock_response.status_code = 200
164 | mock_post.return_value = mock_response
165 |
166 | # Call the method
167 | params = CreateArticleParams(
168 | title="Test Article",
169 | short_description="Test Article",
170 | text="This is a test article content",
171 | knowledge_base="kb001",
172 | category="cat001",
173 | keywords="test,article,knowledge",
174 | article_type="text"
175 | )
176 | result = create_article(self.server_config, self.auth_manager, params)
177 |
178 | # Verify the result
179 | self.assertTrue(result.success)
180 | self.assertEqual("art001", result.article_id)
181 | self.assertEqual("Test Article", result.article_title)
182 |
183 | # Verify the request
184 | mock_post.assert_called_once()
185 | args, kwargs = mock_post.call_args
186 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge", args[0])
187 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
188 | self.assertEqual("Test Article", kwargs["json"]["short_description"])
189 | self.assertEqual("This is a test article content", kwargs["json"]["text"])
190 | self.assertEqual("kb001", kwargs["json"]["kb_knowledge_base"])
191 | self.assertEqual("cat001", kwargs["json"]["kb_category"])
192 | self.assertEqual("text", kwargs["json"]["article_type"])
193 | self.assertEqual("test,article,knowledge", kwargs["json"]["keywords"])
194 |
195 | @patch("servicenow_mcp.tools.knowledge_base.requests.patch")
196 | def test_update_article(self, mock_patch):
197 | """Test updating a knowledge article."""
198 | # Mock response
199 | mock_response = MagicMock()
200 | mock_response.json.return_value = {
201 | "result": {
202 | "sys_id": "art001",
203 | "short_description": "Updated Article",
204 | "text": "This is an updated article content",
205 | "kb_category": "cat002",
206 | "keywords": "updated,article,knowledge",
207 | "workflow_state": "draft",
208 | }
209 | }
210 | mock_response.status_code = 200
211 | mock_patch.return_value = mock_response
212 |
213 | # Call the method
214 | params = UpdateArticleParams(
215 | article_id="art001",
216 | title="Updated Article",
217 | text="This is an updated article content",
218 | category="cat002",
219 | keywords="updated,article,knowledge"
220 | )
221 | result = update_article(self.server_config, self.auth_manager, params)
222 |
223 | # Verify the result
224 | self.assertTrue(result.success)
225 | self.assertEqual("art001", result.article_id)
226 | self.assertEqual("Updated Article", result.article_title)
227 |
228 | # Verify the request
229 | mock_patch.assert_called_once()
230 | args, kwargs = mock_patch.call_args
231 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge/art001", args[0])
232 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
233 | self.assertEqual("Updated Article", kwargs["json"]["short_description"])
234 | self.assertEqual("This is an updated article content", kwargs["json"]["text"])
235 | self.assertEqual("cat002", kwargs["json"]["kb_category"])
236 | self.assertEqual("updated,article,knowledge", kwargs["json"]["keywords"])
237 |
238 | @patch("servicenow_mcp.tools.knowledge_base.requests.patch")
239 | def test_publish_article(self, mock_patch):
240 | """Test publishing a knowledge article."""
241 | # Mock response
242 | mock_response = MagicMock()
243 | mock_response.json.return_value = {
244 | "result": {
245 | "sys_id": "art001",
246 | "short_description": "Test Article",
247 | "workflow_state": "published",
248 | }
249 | }
250 | mock_response.status_code = 200
251 | mock_patch.return_value = mock_response
252 |
253 | # Call the method
254 | params = PublishArticleParams(
255 | article_id="art001",
256 | workflow_state="published"
257 | )
258 | result = publish_article(self.server_config, self.auth_manager, params)
259 |
260 | # Verify the result
261 | self.assertTrue(result.success)
262 | self.assertEqual("art001", result.article_id)
263 | self.assertEqual("Test Article", result.article_title)
264 | self.assertEqual("published", result.workflow_state)
265 |
266 | # Verify the request
267 | mock_patch.assert_called_once()
268 | args, kwargs = mock_patch.call_args
269 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge/art001", args[0])
270 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
271 | self.assertEqual("published", kwargs["json"]["workflow_state"])
272 |
273 | @patch("servicenow_mcp.tools.knowledge_base.requests.get")
274 | def test_list_articles(self, mock_get):
275 | """Test listing knowledge articles."""
276 | # Mock response
277 | mock_response = MagicMock()
278 | mock_response.json.return_value = {
279 | "result": [
280 | {
281 | "sys_id": "art001",
282 | "short_description": "Test Article 1",
283 | "kb_knowledge_base": {"display_value": "IT Knowledge Base"},
284 | "kb_category": {"display_value": "Network"},
285 | "workflow_state": {"display_value": "Published"},
286 | "sys_created_on": "2023-01-01 00:00:00",
287 | "sys_updated_on": "2023-01-02 00:00:00",
288 | },
289 | {
290 | "sys_id": "art002",
291 | "short_description": "Test Article 2",
292 | "kb_knowledge_base": {"display_value": "IT Knowledge Base"},
293 | "kb_category": {"display_value": "Software"},
294 | "workflow_state": {"display_value": "Draft"},
295 | "sys_created_on": "2023-01-03 00:00:00",
296 | "sys_updated_on": "2023-01-04 00:00:00",
297 | }
298 | ]
299 | }
300 | mock_response.status_code = 200
301 | mock_get.return_value = mock_response
302 |
303 | # Call the method
304 | params = ListArticlesParams(
305 | limit=10,
306 | offset=0,
307 | knowledge_base="kb001",
308 | category="cat001",
309 | workflow_state="published",
310 | query="network"
311 | )
312 | result = list_articles(self.server_config, self.auth_manager, params)
313 |
314 | # Verify the result
315 | self.assertTrue(result["success"])
316 | self.assertEqual(2, len(result["articles"]))
317 | self.assertEqual("art001", result["articles"][0]["id"])
318 | self.assertEqual("Test Article 1", result["articles"][0]["title"])
319 | self.assertEqual("IT Knowledge Base", result["articles"][0]["knowledge_base"])
320 | self.assertEqual("Network", result["articles"][0]["category"])
321 | self.assertEqual("Published", result["articles"][0]["workflow_state"])
322 |
323 | # Verify the request
324 | mock_get.assert_called_once()
325 | args, kwargs = mock_get.call_args
326 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge", args[0])
327 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
328 | self.assertEqual(10, kwargs["params"]["sysparm_limit"])
329 | self.assertEqual(0, kwargs["params"]["sysparm_offset"])
330 | self.assertEqual("all", kwargs["params"]["sysparm_display_value"])
331 |
332 | # Verify the query syntax contains the correct pattern
333 | self.assertIn("sysparm_query", kwargs["params"])
334 | query = kwargs["params"]["sysparm_query"]
335 | self.assertIn("kb_knowledge_base.sys_id=kb001", query)
336 | self.assertIn("kb_category.sys_id=cat001", query)
337 |
338 | @patch("servicenow_mcp.tools.knowledge_base.requests.get")
339 | def test_get_article(self, mock_get):
340 | """Test getting a knowledge article."""
341 | # Mock response
342 | mock_response = MagicMock()
343 | mock_response.json.return_value = {
344 | "result": {
345 | "sys_id": "art001",
346 | "short_description": "Test Article",
347 | "text": "This is a test article content",
348 | "kb_knowledge_base": {"display_value": "IT Knowledge Base"},
349 | "kb_category": {"display_value": "Network"},
350 | "workflow_state": {"display_value": "Published"},
351 | "sys_created_on": "2023-01-01 00:00:00",
352 | "sys_updated_on": "2023-01-02 00:00:00",
353 | "author": {"display_value": "admin"},
354 | "keywords": "test,article,knowledge",
355 | "article_type": "text",
356 | "view_count": "42"
357 | }
358 | }
359 | mock_response.status_code = 200
360 | mock_get.return_value = mock_response
361 |
362 | # Call the method
363 | params = GetArticleParams(article_id="art001")
364 | result = get_article(self.server_config, self.auth_manager, params)
365 |
366 | # Verify the result
367 | self.assertTrue(result["success"])
368 | self.assertEqual("art001", result["article"]["id"])
369 | self.assertEqual("Test Article", result["article"]["title"])
370 | self.assertEqual("This is a test article content", result["article"]["text"])
371 | self.assertEqual("IT Knowledge Base", result["article"]["knowledge_base"])
372 | self.assertEqual("Network", result["article"]["category"])
373 | self.assertEqual("Published", result["article"]["workflow_state"])
374 | self.assertEqual("admin", result["article"]["author"])
375 | self.assertEqual("test,article,knowledge", result["article"]["keywords"])
376 | self.assertEqual("text", result["article"]["article_type"])
377 | self.assertEqual("42", result["article"]["views"])
378 |
379 | # Verify the request
380 | mock_get.assert_called_once()
381 | args, kwargs = mock_get.call_args
382 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge/art001", args[0])
383 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
384 | self.assertEqual("true", kwargs["params"]["sysparm_display_value"])
385 |
386 | @patch("servicenow_mcp.tools.knowledge_base.requests.post")
387 | def test_create_knowledge_base_error(self, mock_post):
388 | """Test error handling when creating a knowledge base."""
389 | # Mock error response
390 | mock_post.side_effect = requests.RequestException("API error")
391 |
392 | # Call the method
393 | params = CreateKnowledgeBaseParams(title="Test Knowledge Base")
394 | result = create_knowledge_base(self.server_config, self.auth_manager, params)
395 |
396 | # Verify the result
397 | self.assertFalse(result.success)
398 | self.assertIn("Failed to create knowledge base", result.message)
399 |
400 | @patch("servicenow_mcp.tools.knowledge_base.requests.get")
401 | def test_get_article_not_found(self, mock_get):
402 | """Test getting a non-existent article."""
403 | # Mock empty response
404 | mock_response = MagicMock()
405 | mock_response.json.return_value = {"result": {}}
406 | mock_response.status_code = 200
407 | mock_get.return_value = mock_response
408 |
409 | # Call the method
410 | params = GetArticleParams(article_id="nonexistent")
411 | result = get_article(self.server_config, self.auth_manager, params)
412 |
413 | # Verify the result
414 | self.assertFalse(result["success"])
415 | self.assertIn("not found", result["message"])
416 |
417 | @patch("servicenow_mcp.tools.knowledge_base.requests.get")
418 | def test_list_knowledge_bases(self, mock_get):
419 | """Test listing knowledge bases."""
420 | # Mock response
421 | mock_response = MagicMock()
422 | mock_response.json.return_value = {
423 | "result": [
424 | {
425 | "sys_id": "kb001",
426 | "title": "IT Knowledge Base",
427 | "description": "Knowledge base for IT resources",
428 | "owner": {"display_value": "admin"},
429 | "kb_managers": {"display_value": "it_managers"},
430 | "active": "true",
431 | "sys_created_on": "2023-01-01 00:00:00",
432 | "sys_updated_on": "2023-01-02 00:00:00",
433 | },
434 | {
435 | "sys_id": "kb002",
436 | "title": "HR Knowledge Base",
437 | "description": "Knowledge base for HR resources",
438 | "owner": {"display_value": "hr_admin"},
439 | "kb_managers": {"display_value": "hr_managers"},
440 | "active": "true",
441 | "sys_created_on": "2023-01-03 00:00:00",
442 | "sys_updated_on": "2023-01-04 00:00:00",
443 | }
444 | ]
445 | }
446 | mock_response.status_code = 200
447 | mock_get.return_value = mock_response
448 |
449 | # Call the method
450 | params = ListKnowledgeBasesParams(
451 | limit=10,
452 | offset=0,
453 | active=True,
454 | query="IT"
455 | )
456 | result = list_knowledge_bases(self.server_config, self.auth_manager, params)
457 |
458 | # Verify the result
459 | self.assertTrue(result["success"])
460 | self.assertEqual(2, len(result["knowledge_bases"]))
461 | self.assertEqual("kb001", result["knowledge_bases"][0]["id"])
462 | self.assertEqual("IT Knowledge Base", result["knowledge_bases"][0]["title"])
463 | self.assertEqual("Knowledge base for IT resources", result["knowledge_bases"][0]["description"])
464 | self.assertEqual("admin", result["knowledge_bases"][0]["owner"])
465 | self.assertEqual("it_managers", result["knowledge_bases"][0]["managers"])
466 | self.assertTrue(result["knowledge_bases"][0]["active"])
467 |
468 | # Verify the request
469 | mock_get.assert_called_once()
470 | args, kwargs = mock_get.call_args
471 | self.assertEqual(f"{self.server_config.api_url}/table/kb_knowledge_base", args[0])
472 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
473 | self.assertEqual(10, kwargs["params"]["sysparm_limit"])
474 | self.assertEqual(0, kwargs["params"]["sysparm_offset"])
475 | self.assertEqual("true", kwargs["params"]["sysparm_display_value"])
476 | self.assertEqual("active=true^titleLIKEIT^ORdescriptionLIKEIT", kwargs["params"]["sysparm_query"])
477 |
478 | @patch("servicenow_mcp.tools.knowledge_base.requests.get")
479 | def test_list_categories(self, mock_get):
480 | """Test listing categories in a knowledge base."""
481 | # Mock response
482 | mock_response = MagicMock()
483 | mock_response.json.return_value = {
484 | "result": [
485 | {
486 | "sys_id": "cat001",
487 | "label": "Network Troubleshooting",
488 | "description": "Articles for network troubleshooting",
489 | "kb_knowledge_base": {"display_value": "IT Knowledge Base"},
490 | "parent": {"display_value": ""},
491 | "active": "true",
492 | "sys_created_on": "2023-01-01 00:00:00",
493 | "sys_updated_on": "2023-01-02 00:00:00",
494 | },
495 | {
496 | "sys_id": "cat002",
497 | "label": "Software Setup",
498 | "description": "Articles for software installation",
499 | "kb_knowledge_base": {"display_value": "IT Knowledge Base"},
500 | "parent": {"display_value": ""},
501 | "active": "true",
502 | "sys_created_on": "2023-01-03 00:00:00",
503 | "sys_updated_on": "2023-01-04 00:00:00",
504 | }
505 | ]
506 | }
507 | mock_response.status_code = 200
508 | mock_get.return_value = mock_response
509 |
510 | # Call the method
511 | params = ListCategoriesParams(
512 | knowledge_base="kb001",
513 | active=True,
514 | query="Network"
515 | )
516 | result = list_categories(self.server_config, self.auth_manager, params)
517 |
518 | # Verify the result
519 | self.assertTrue(result["success"])
520 | self.assertEqual(2, len(result["categories"]))
521 | self.assertEqual("cat001", result["categories"][0]["id"])
522 | self.assertEqual("Network Troubleshooting", result["categories"][0]["title"])
523 | self.assertEqual("Articles for network troubleshooting", result["categories"][0]["description"])
524 | self.assertEqual("IT Knowledge Base", result["categories"][0]["knowledge_base"])
525 | self.assertEqual("", result["categories"][0]["parent_category"])
526 | self.assertTrue(result["categories"][0]["active"])
527 |
528 | # Verify the request
529 | mock_get.assert_called_once()
530 | args, kwargs = mock_get.call_args
531 | self.assertEqual(f"{self.server_config.api_url}/table/kb_category", args[0])
532 | self.assertEqual(self.auth_manager.get_headers(), kwargs["headers"])
533 | self.assertEqual(10, kwargs["params"]["sysparm_limit"])
534 | self.assertEqual(0, kwargs["params"]["sysparm_offset"])
535 | self.assertEqual("all", kwargs["params"]["sysparm_display_value"])
536 |
537 | # Verify the query syntax contains the correct pattern
538 | self.assertIn("sysparm_query", kwargs["params"])
539 | query = kwargs["params"]["sysparm_query"]
540 | self.assertIn("kb_knowledge_base.sys_id=kb001", query)
541 | self.assertIn("active=true", query)
542 | self.assertIn("labelLIKENetwork", query)
543 |
544 |
545 | class TestKnowledgeBaseParams(unittest.TestCase):
546 | """Tests for the knowledge base parameter classes."""
547 |
548 | def test_create_knowledge_base_params(self):
549 | """Test CreateKnowledgeBaseParams validation."""
550 | # Minimal required parameters
551 | params = CreateKnowledgeBaseParams(title="Test Knowledge Base")
552 | self.assertEqual("Test Knowledge Base", params.title)
553 | self.assertEqual("Knowledge - Instant Publish", params.publish_workflow)
554 |
555 | # All parameters
556 | params = CreateKnowledgeBaseParams(
557 | title="Test Knowledge Base",
558 | description="Test Description",
559 | owner="admin",
560 | managers="it_managers",
561 | publish_workflow="Custom Workflow",
562 | retire_workflow="Custom Retire Workflow"
563 | )
564 | self.assertEqual("Test Knowledge Base", params.title)
565 | self.assertEqual("Test Description", params.description)
566 | self.assertEqual("admin", params.owner)
567 | self.assertEqual("it_managers", params.managers)
568 | self.assertEqual("Custom Workflow", params.publish_workflow)
569 | self.assertEqual("Custom Retire Workflow", params.retire_workflow)
570 |
571 | def test_create_category_params(self):
572 | """Test CreateCategoryParams validation."""
573 | # Required parameters
574 | params = CreateCategoryParams(
575 | title="Test Category",
576 | knowledge_base="kb001"
577 | )
578 | self.assertEqual("Test Category", params.title)
579 | self.assertEqual("kb001", params.knowledge_base)
580 | self.assertTrue(params.active)
581 |
582 | # All parameters
583 | params = CreateCategoryParams(
584 | title="Test Category",
585 | description="Test Description",
586 | knowledge_base="kb001",
587 | parent_category="parent001",
588 | active=False
589 | )
590 | self.assertEqual("Test Category", params.title)
591 | self.assertEqual("Test Description", params.description)
592 | self.assertEqual("kb001", params.knowledge_base)
593 | self.assertEqual("parent001", params.parent_category)
594 | self.assertFalse(params.active)
595 |
596 | def test_create_article_params(self):
597 | """Test CreateArticleParams validation."""
598 | # Required parameters
599 | params = CreateArticleParams(
600 | title="Test Article",
601 | text="Test content",
602 | short_description="Test short description",
603 | knowledge_base="kb001",
604 | category="cat001"
605 | )
606 | self.assertEqual("Test Article", params.title)
607 | self.assertEqual("Test content", params.text)
608 | self.assertEqual("Test short description", params.short_description)
609 | self.assertEqual("kb001", params.knowledge_base)
610 | self.assertEqual("cat001", params.category)
611 | self.assertEqual("text", params.article_type)
612 |
613 | # All parameters
614 | params = CreateArticleParams(
615 | title="Test Article",
616 | text="Test content",
617 | short_description="Test short description",
618 | knowledge_base="kb001",
619 | category="cat001",
620 | keywords="test,article",
621 | article_type="html"
622 | )
623 | self.assertEqual("Test Article", params.title)
624 | self.assertEqual("Test content", params.text)
625 | self.assertEqual("Test short description", params.short_description)
626 | self.assertEqual("kb001", params.knowledge_base)
627 | self.assertEqual("cat001", params.category)
628 | self.assertEqual("test,article", params.keywords)
629 | self.assertEqual("html", params.article_type)
630 |
631 |
632 | if __name__ == "__main__":
633 | unittest.main()
```