This is page 5 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
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/user_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | User management tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing users and groups in ServiceNow.
5 | """
6 |
7 | import logging
8 | from typing import 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 CreateUserParams(BaseModel):
20 | """Parameters for creating a user."""
21 |
22 | user_name: str = Field(..., description="Username for the user")
23 | first_name: str = Field(..., description="First name of the user")
24 | last_name: str = Field(..., description="Last name of the user")
25 | email: str = Field(..., description="Email address of the user")
26 | title: Optional[str] = Field(None, description="Job title of the user")
27 | department: Optional[str] = Field(None, description="Department the user belongs to")
28 | manager: Optional[str] = Field(None, description="Manager of the user (sys_id or username)")
29 | roles: Optional[List[str]] = Field(None, description="Roles to assign to the user")
30 | phone: Optional[str] = Field(None, description="Phone number of the user")
31 | mobile_phone: Optional[str] = Field(None, description="Mobile phone number of the user")
32 | location: Optional[str] = Field(None, description="Location of the user")
33 | password: Optional[str] = Field(None, description="Password for the user account")
34 | active: Optional[bool] = Field(True, description="Whether the user account is active")
35 |
36 |
37 | class UpdateUserParams(BaseModel):
38 | """Parameters for updating a user."""
39 |
40 | user_id: str = Field(..., description="User ID or sys_id to update")
41 | user_name: Optional[str] = Field(None, description="Username for the user")
42 | first_name: Optional[str] = Field(None, description="First name of the user")
43 | last_name: Optional[str] = Field(None, description="Last name of the user")
44 | email: Optional[str] = Field(None, description="Email address of the user")
45 | title: Optional[str] = Field(None, description="Job title of the user")
46 | department: Optional[str] = Field(None, description="Department the user belongs to")
47 | manager: Optional[str] = Field(None, description="Manager of the user (sys_id or username)")
48 | roles: Optional[List[str]] = Field(None, description="Roles to assign to the user")
49 | phone: Optional[str] = Field(None, description="Phone number of the user")
50 | mobile_phone: Optional[str] = Field(None, description="Mobile phone number of the user")
51 | location: Optional[str] = Field(None, description="Location of the user")
52 | password: Optional[str] = Field(None, description="Password for the user account")
53 | active: Optional[bool] = Field(None, description="Whether the user account is active")
54 |
55 |
56 | class GetUserParams(BaseModel):
57 | """Parameters for getting a user."""
58 |
59 | user_id: Optional[str] = Field(None, description="User ID or sys_id")
60 | user_name: Optional[str] = Field(None, description="Username of the user")
61 | email: Optional[str] = Field(None, description="Email address of the user")
62 |
63 |
64 | class ListUsersParams(BaseModel):
65 | """Parameters for listing users."""
66 |
67 | limit: int = Field(10, description="Maximum number of users to return")
68 | offset: int = Field(0, description="Offset for pagination")
69 | active: Optional[bool] = Field(None, description="Filter by active status")
70 | department: Optional[str] = Field(None, description="Filter by department")
71 | query: Optional[str] = Field(
72 | None,
73 | description="Case-insensitive search term that matches against name, username, or email fields. Uses ServiceNow's LIKE operator for partial matching.",
74 | )
75 |
76 |
77 | class CreateGroupParams(BaseModel):
78 | """Parameters for creating a group."""
79 |
80 | name: str = Field(..., description="Name of the group")
81 | description: Optional[str] = Field(None, description="Description of the group")
82 | manager: Optional[str] = Field(None, description="Manager of the group (sys_id or username)")
83 | parent: Optional[str] = Field(None, description="Parent group (sys_id or name)")
84 | type: Optional[str] = Field(None, description="Type of the group")
85 | email: Optional[str] = Field(None, description="Email address for the group")
86 | members: Optional[List[str]] = Field(
87 | None, description="List of user sys_ids or usernames to add as members"
88 | )
89 | active: Optional[bool] = Field(True, description="Whether the group is active")
90 |
91 |
92 | class UpdateGroupParams(BaseModel):
93 | """Parameters for updating a group."""
94 |
95 | group_id: str = Field(..., description="Group ID or sys_id to update")
96 | name: Optional[str] = Field(None, description="Name of the group")
97 | description: Optional[str] = Field(None, description="Description of the group")
98 | manager: Optional[str] = Field(None, description="Manager of the group (sys_id or username)")
99 | parent: Optional[str] = Field(None, description="Parent group (sys_id or name)")
100 | type: Optional[str] = Field(None, description="Type of the group")
101 | email: Optional[str] = Field(None, description="Email address for the group")
102 | active: Optional[bool] = Field(None, description="Whether the group is active")
103 |
104 |
105 | class AddGroupMembersParams(BaseModel):
106 | """Parameters for adding members to a group."""
107 |
108 | group_id: str = Field(..., description="Group ID or sys_id")
109 | members: List[str] = Field(
110 | ..., description="List of user sys_ids or usernames to add as members"
111 | )
112 |
113 |
114 | class RemoveGroupMembersParams(BaseModel):
115 | """Parameters for removing members from a group."""
116 |
117 | group_id: str = Field(..., description="Group ID or sys_id")
118 | members: List[str] = Field(
119 | ..., description="List of user sys_ids or usernames to remove as members"
120 | )
121 |
122 |
123 | class ListGroupsParams(BaseModel):
124 | """Parameters for listing groups."""
125 |
126 | limit: int = Field(10, description="Maximum number of groups to return")
127 | offset: int = Field(0, description="Offset for pagination")
128 | active: Optional[bool] = Field(None, description="Filter by active status")
129 | query: Optional[str] = Field(
130 | None,
131 | description="Case-insensitive search term that matches against group name or description fields. Uses ServiceNow's LIKE operator for partial matching.",
132 | )
133 | type: Optional[str] = Field(None, description="Filter by group type")
134 |
135 |
136 | class UserResponse(BaseModel):
137 | """Response from user operations."""
138 |
139 | success: bool = Field(..., description="Whether the operation was successful")
140 | message: str = Field(..., description="Message describing the result")
141 | user_id: Optional[str] = Field(None, description="ID of the affected user")
142 | user_name: Optional[str] = Field(None, description="Username of the affected user")
143 |
144 |
145 | class GroupResponse(BaseModel):
146 | """Response from group operations."""
147 |
148 | success: bool = Field(..., description="Whether the operation was successful")
149 | message: str = Field(..., description="Message describing the result")
150 | group_id: Optional[str] = Field(None, description="ID of the affected group")
151 | group_name: Optional[str] = Field(None, description="Name of the affected group")
152 |
153 |
154 | def create_user(
155 | config: ServerConfig,
156 | auth_manager: AuthManager,
157 | params: CreateUserParams,
158 | ) -> UserResponse:
159 | """
160 | Create a new user in ServiceNow.
161 |
162 | Args:
163 | config: Server configuration.
164 | auth_manager: Authentication manager.
165 | params: Parameters for creating the user.
166 |
167 | Returns:
168 | Response with the created user details.
169 | """
170 | api_url = f"{config.api_url}/table/sys_user"
171 |
172 | # Build request data
173 | data = {
174 | "user_name": params.user_name,
175 | "first_name": params.first_name,
176 | "last_name": params.last_name,
177 | "email": params.email,
178 | "active": str(params.active).lower(),
179 | }
180 |
181 | if params.title:
182 | data["title"] = params.title
183 | if params.department:
184 | data["department"] = params.department
185 | if params.manager:
186 | data["manager"] = params.manager
187 | if params.phone:
188 | data["phone"] = params.phone
189 | if params.mobile_phone:
190 | data["mobile_phone"] = params.mobile_phone
191 | if params.location:
192 | data["location"] = params.location
193 | if params.password:
194 | data["user_password"] = params.password
195 |
196 | # Make request
197 | try:
198 | response = requests.post(
199 | api_url,
200 | json=data,
201 | headers=auth_manager.get_headers(),
202 | timeout=config.timeout,
203 | )
204 | response.raise_for_status()
205 |
206 | result = response.json().get("result", {})
207 |
208 | # Handle role assignments if provided
209 | if params.roles and result.get("sys_id"):
210 | assign_roles_to_user(config, auth_manager, result.get("sys_id"), params.roles)
211 |
212 | return UserResponse(
213 | success=True,
214 | message="User created successfully",
215 | user_id=result.get("sys_id"),
216 | user_name=result.get("user_name"),
217 | )
218 |
219 | except requests.RequestException as e:
220 | logger.error(f"Failed to create user: {e}")
221 | return UserResponse(
222 | success=False,
223 | message=f"Failed to create user: {str(e)}",
224 | )
225 |
226 |
227 | def update_user(
228 | config: ServerConfig,
229 | auth_manager: AuthManager,
230 | params: UpdateUserParams,
231 | ) -> UserResponse:
232 | """
233 | Update an existing user in ServiceNow.
234 |
235 | Args:
236 | config: Server configuration.
237 | auth_manager: Authentication manager.
238 | params: Parameters for updating the user.
239 |
240 | Returns:
241 | Response with the updated user details.
242 | """
243 | api_url = f"{config.api_url}/table/sys_user/{params.user_id}"
244 |
245 | # Build request data
246 | data = {}
247 | if params.user_name:
248 | data["user_name"] = params.user_name
249 | if params.first_name:
250 | data["first_name"] = params.first_name
251 | if params.last_name:
252 | data["last_name"] = params.last_name
253 | if params.email:
254 | data["email"] = params.email
255 | if params.title:
256 | data["title"] = params.title
257 | if params.department:
258 | data["department"] = params.department
259 | if params.manager:
260 | data["manager"] = params.manager
261 | if params.phone:
262 | data["phone"] = params.phone
263 | if params.mobile_phone:
264 | data["mobile_phone"] = params.mobile_phone
265 | if params.location:
266 | data["location"] = params.location
267 | if params.password:
268 | data["user_password"] = params.password
269 | if params.active is not None:
270 | data["active"] = str(params.active).lower()
271 |
272 | # Make request
273 | try:
274 | response = requests.patch(
275 | api_url,
276 | json=data,
277 | headers=auth_manager.get_headers(),
278 | timeout=config.timeout,
279 | )
280 | response.raise_for_status()
281 |
282 | result = response.json().get("result", {})
283 |
284 | # Handle role assignments if provided
285 | if params.roles:
286 | assign_roles_to_user(config, auth_manager, params.user_id, params.roles)
287 |
288 | return UserResponse(
289 | success=True,
290 | message="User updated successfully",
291 | user_id=result.get("sys_id"),
292 | user_name=result.get("user_name"),
293 | )
294 |
295 | except requests.RequestException as e:
296 | logger.error(f"Failed to update user: {e}")
297 | return UserResponse(
298 | success=False,
299 | message=f"Failed to update user: {str(e)}",
300 | )
301 |
302 |
303 | def get_user(
304 | config: ServerConfig,
305 | auth_manager: AuthManager,
306 | params: GetUserParams,
307 | ) -> dict:
308 | """
309 | Get a user from ServiceNow.
310 |
311 | Args:
312 | config: Server configuration.
313 | auth_manager: Authentication manager.
314 | params: Parameters for getting the user.
315 |
316 | Returns:
317 | Dictionary containing user details.
318 | """
319 | api_url = f"{config.api_url}/table/sys_user"
320 | query_params = {}
321 |
322 | # Build query parameters
323 | if params.user_id:
324 | query_params["sysparm_query"] = f"sys_id={params.user_id}"
325 | elif params.user_name:
326 | query_params["sysparm_query"] = f"user_name={params.user_name}"
327 | elif params.email:
328 | query_params["sysparm_query"] = f"email={params.email}"
329 | else:
330 | return {"success": False, "message": "At least one search parameter is required"}
331 |
332 | query_params["sysparm_limit"] = "1"
333 | query_params["sysparm_display_value"] = "true"
334 |
335 | # Make request
336 | try:
337 | response = requests.get(
338 | api_url,
339 | params=query_params,
340 | headers=auth_manager.get_headers(),
341 | timeout=config.timeout,
342 | )
343 | response.raise_for_status()
344 |
345 | result = response.json().get("result", [])
346 | if not result:
347 | return {"success": False, "message": "User not found"}
348 |
349 | return {"success": True, "message": "User found", "user": result[0]}
350 |
351 | except requests.RequestException as e:
352 | logger.error(f"Failed to get user: {e}")
353 | return {"success": False, "message": f"Failed to get user: {str(e)}"}
354 |
355 |
356 | def list_users(
357 | config: ServerConfig,
358 | auth_manager: AuthManager,
359 | params: ListUsersParams,
360 | ) -> dict:
361 | """
362 | List users from ServiceNow.
363 |
364 | Args:
365 | config: Server configuration.
366 | auth_manager: Authentication manager.
367 | params: Parameters for listing users.
368 |
369 | Returns:
370 | Dictionary containing list of users.
371 | """
372 | api_url = f"{config.api_url}/table/sys_user"
373 | query_params = {
374 | "sysparm_limit": str(params.limit),
375 | "sysparm_offset": str(params.offset),
376 | "sysparm_display_value": "true",
377 | }
378 |
379 | # Build query
380 | query_parts = []
381 | if params.active is not None:
382 | query_parts.append(f"active={str(params.active).lower()}")
383 | if params.department:
384 | query_parts.append(f"department={params.department}")
385 | if params.query:
386 | query_parts.append(
387 | f"^nameLIKE{params.query}^ORuser_nameLIKE{params.query}^ORemailLIKE{params.query}"
388 | )
389 |
390 | if query_parts:
391 | query_params["sysparm_query"] = "^".join(query_parts)
392 |
393 | # Make request
394 | try:
395 | response = requests.get(
396 | api_url,
397 | params=query_params,
398 | headers=auth_manager.get_headers(),
399 | timeout=config.timeout,
400 | )
401 | response.raise_for_status()
402 |
403 | result = response.json().get("result", [])
404 |
405 | return {
406 | "success": True,
407 | "message": f"Found {len(result)} users",
408 | "users": result,
409 | "count": len(result),
410 | }
411 |
412 | except requests.RequestException as e:
413 | logger.error(f"Failed to list users: {e}")
414 | return {"success": False, "message": f"Failed to list users: {str(e)}"}
415 |
416 |
417 | def list_groups(
418 | config: ServerConfig,
419 | auth_manager: AuthManager,
420 | params: ListGroupsParams,
421 | ) -> dict:
422 | """
423 | List groups from ServiceNow.
424 |
425 | Args:
426 | config: Server configuration.
427 | auth_manager: Authentication manager.
428 | params: Parameters for listing groups.
429 |
430 | Returns:
431 | Dictionary containing list of groups.
432 | """
433 | api_url = f"{config.api_url}/table/sys_user_group"
434 | query_params = {
435 | "sysparm_limit": str(params.limit),
436 | "sysparm_offset": str(params.offset),
437 | "sysparm_display_value": "true",
438 | }
439 |
440 | # Build query
441 | query_parts = []
442 | if params.active is not None:
443 | query_parts.append(f"active={str(params.active).lower()}")
444 | if params.type:
445 | query_parts.append(f"type={params.type}")
446 | if params.query:
447 | query_parts.append(f"^nameLIKE{params.query}^ORdescriptionLIKE{params.query}")
448 |
449 | if query_parts:
450 | query_params["sysparm_query"] = "^".join(query_parts)
451 |
452 | # Make request
453 | try:
454 | response = requests.get(
455 | api_url,
456 | params=query_params,
457 | headers=auth_manager.get_headers(),
458 | timeout=config.timeout,
459 | )
460 | response.raise_for_status()
461 |
462 | result = response.json().get("result", [])
463 |
464 | return {
465 | "success": True,
466 | "message": f"Found {len(result)} groups",
467 | "groups": result,
468 | "count": len(result),
469 | }
470 |
471 | except requests.RequestException as e:
472 | logger.error(f"Failed to list groups: {e}")
473 | return {"success": False, "message": f"Failed to list groups: {str(e)}"}
474 |
475 |
476 | def assign_roles_to_user(
477 | config: ServerConfig,
478 | auth_manager: AuthManager,
479 | user_id: str,
480 | roles: List[str],
481 | ) -> bool:
482 | """
483 | Assign roles to a user in ServiceNow.
484 |
485 | Args:
486 | config: Server configuration.
487 | auth_manager: Authentication manager.
488 | user_id: User ID or sys_id.
489 | roles: List of roles to assign.
490 |
491 | Returns:
492 | Boolean indicating success.
493 | """
494 | # For each role, create a user_role record
495 | api_url = f"{config.api_url}/table/sys_user_has_role"
496 |
497 | success = True
498 | for role in roles:
499 | # First check if the role exists
500 | role_id = get_role_id(config, auth_manager, role)
501 | if not role_id:
502 | logger.warning(f"Role '{role}' not found, skipping assignment")
503 | continue
504 |
505 | # Check if the user already has this role
506 | if check_user_has_role(config, auth_manager, user_id, role_id):
507 | logger.info(f"User already has role '{role}', skipping assignment")
508 | continue
509 |
510 | # Create the user role assignment
511 | data = {
512 | "user": user_id,
513 | "role": role_id,
514 | }
515 |
516 | try:
517 | response = requests.post(
518 | api_url,
519 | json=data,
520 | headers=auth_manager.get_headers(),
521 | timeout=config.timeout,
522 | )
523 | response.raise_for_status()
524 | except requests.RequestException as e:
525 | logger.error(f"Failed to assign role '{role}' to user: {e}")
526 | success = False
527 |
528 | return success
529 |
530 |
531 | def get_role_id(
532 | config: ServerConfig,
533 | auth_manager: AuthManager,
534 | role_name: str,
535 | ) -> Optional[str]:
536 | """
537 | Get the sys_id of a role by its name.
538 |
539 | Args:
540 | config: Server configuration.
541 | auth_manager: Authentication manager.
542 | role_name: Name of the role.
543 |
544 | Returns:
545 | sys_id of the role if found, None otherwise.
546 | """
547 | api_url = f"{config.api_url}/table/sys_user_role"
548 | query_params = {
549 | "sysparm_query": f"name={role_name}",
550 | "sysparm_limit": "1",
551 | }
552 |
553 | try:
554 | response = requests.get(
555 | api_url,
556 | params=query_params,
557 | headers=auth_manager.get_headers(),
558 | timeout=config.timeout,
559 | )
560 | response.raise_for_status()
561 |
562 | result = response.json().get("result", [])
563 | if not result:
564 | return None
565 |
566 | return result[0].get("sys_id")
567 |
568 | except requests.RequestException as e:
569 | logger.error(f"Failed to get role ID: {e}")
570 | return None
571 |
572 |
573 | def check_user_has_role(
574 | config: ServerConfig,
575 | auth_manager: AuthManager,
576 | user_id: str,
577 | role_id: str,
578 | ) -> bool:
579 | """
580 | Check if a user has a specific role.
581 |
582 | Args:
583 | config: Server configuration.
584 | auth_manager: Authentication manager.
585 | user_id: User ID or sys_id.
586 | role_id: Role ID or sys_id.
587 |
588 | Returns:
589 | Boolean indicating whether the user has the role.
590 | """
591 | api_url = f"{config.api_url}/table/sys_user_has_role"
592 | query_params = {
593 | "sysparm_query": f"user={user_id}^role={role_id}",
594 | "sysparm_limit": "1",
595 | }
596 |
597 | try:
598 | response = requests.get(
599 | api_url,
600 | params=query_params,
601 | headers=auth_manager.get_headers(),
602 | timeout=config.timeout,
603 | )
604 | response.raise_for_status()
605 |
606 | result = response.json().get("result", [])
607 | return len(result) > 0
608 |
609 | except requests.RequestException as e:
610 | logger.error(f"Failed to check if user has role: {e}")
611 | return False
612 |
613 |
614 | def create_group(
615 | config: ServerConfig,
616 | auth_manager: AuthManager,
617 | params: CreateGroupParams,
618 | ) -> GroupResponse:
619 | """
620 | Create a new group in ServiceNow.
621 |
622 | Args:
623 | config: Server configuration.
624 | auth_manager: Authentication manager.
625 | params: Parameters for creating the group.
626 |
627 | Returns:
628 | Response with the created group details.
629 | """
630 | api_url = f"{config.api_url}/table/sys_user_group"
631 |
632 | # Build request data
633 | data = {
634 | "name": params.name,
635 | "active": str(params.active).lower(),
636 | }
637 |
638 | if params.description:
639 | data["description"] = params.description
640 | if params.manager:
641 | data["manager"] = params.manager
642 | if params.parent:
643 | data["parent"] = params.parent
644 | if params.type:
645 | data["type"] = params.type
646 | if params.email:
647 | data["email"] = params.email
648 |
649 | # Make request
650 | try:
651 | response = requests.post(
652 | api_url,
653 | json=data,
654 | headers=auth_manager.get_headers(),
655 | timeout=config.timeout,
656 | )
657 | response.raise_for_status()
658 |
659 | result = response.json().get("result", {})
660 | group_id = result.get("sys_id")
661 |
662 | # Add members if provided
663 | if params.members and group_id:
664 | add_group_members(
665 | config,
666 | auth_manager,
667 | AddGroupMembersParams(group_id=group_id, members=params.members),
668 | )
669 |
670 | return GroupResponse(
671 | success=True,
672 | message="Group created successfully",
673 | group_id=group_id,
674 | group_name=result.get("name"),
675 | )
676 |
677 | except requests.RequestException as e:
678 | logger.error(f"Failed to create group: {e}")
679 | return GroupResponse(
680 | success=False,
681 | message=f"Failed to create group: {str(e)}",
682 | )
683 |
684 |
685 | def update_group(
686 | config: ServerConfig,
687 | auth_manager: AuthManager,
688 | params: UpdateGroupParams,
689 | ) -> GroupResponse:
690 | """
691 | Update an existing group in ServiceNow.
692 |
693 | Args:
694 | config: Server configuration.
695 | auth_manager: Authentication manager.
696 | params: Parameters for updating the group.
697 |
698 | Returns:
699 | Response with the updated group details.
700 | """
701 | api_url = f"{config.api_url}/table/sys_user_group/{params.group_id}"
702 |
703 | # Build request data
704 | data = {}
705 | if params.name:
706 | data["name"] = params.name
707 | if params.description:
708 | data["description"] = params.description
709 | if params.manager:
710 | data["manager"] = params.manager
711 | if params.parent:
712 | data["parent"] = params.parent
713 | if params.type:
714 | data["type"] = params.type
715 | if params.email:
716 | data["email"] = params.email
717 | if params.active is not None:
718 | data["active"] = str(params.active).lower()
719 |
720 | # Make request
721 | try:
722 | response = requests.patch(
723 | api_url,
724 | json=data,
725 | headers=auth_manager.get_headers(),
726 | timeout=config.timeout,
727 | )
728 | response.raise_for_status()
729 |
730 | result = response.json().get("result", {})
731 |
732 | return GroupResponse(
733 | success=True,
734 | message="Group updated successfully",
735 | group_id=result.get("sys_id"),
736 | group_name=result.get("name"),
737 | )
738 |
739 | except requests.RequestException as e:
740 | logger.error(f"Failed to update group: {e}")
741 | return GroupResponse(
742 | success=False,
743 | message=f"Failed to update group: {str(e)}",
744 | )
745 |
746 |
747 | def add_group_members(
748 | config: ServerConfig,
749 | auth_manager: AuthManager,
750 | params: AddGroupMembersParams,
751 | ) -> GroupResponse:
752 | """
753 | Add members to a group in ServiceNow.
754 |
755 | Args:
756 | config: Server configuration.
757 | auth_manager: Authentication manager.
758 | params: Parameters for adding members to the group.
759 |
760 | Returns:
761 | Response with the result of the operation.
762 | """
763 | api_url = f"{config.api_url}/table/sys_user_grmember"
764 |
765 | success = True
766 | failed_members = []
767 |
768 | for member in params.members:
769 | # Get user ID if username is provided
770 | user_id = member
771 | if not member.startswith("sys_id:"):
772 | user = get_user(config, auth_manager, GetUserParams(user_name=member))
773 | if not user.get("success"):
774 | user = get_user(config, auth_manager, GetUserParams(email=member))
775 |
776 | if user.get("success"):
777 | user_id = user.get("user", {}).get("sys_id")
778 | else:
779 | success = False
780 | failed_members.append(member)
781 | continue
782 |
783 | # Create group membership
784 | data = {
785 | "group": params.group_id,
786 | "user": user_id,
787 | }
788 |
789 | try:
790 | response = requests.post(
791 | api_url,
792 | json=data,
793 | headers=auth_manager.get_headers(),
794 | timeout=config.timeout,
795 | )
796 | response.raise_for_status()
797 | except requests.RequestException as e:
798 | logger.error(f"Failed to add member '{member}' to group: {e}")
799 | success = False
800 | failed_members.append(member)
801 |
802 | if failed_members:
803 | message = f"Some members could not be added to the group: {', '.join(failed_members)}"
804 | else:
805 | message = "All members added to the group successfully"
806 |
807 | return GroupResponse(
808 | success=success,
809 | message=message,
810 | group_id=params.group_id,
811 | )
812 |
813 |
814 | def remove_group_members(
815 | config: ServerConfig,
816 | auth_manager: AuthManager,
817 | params: RemoveGroupMembersParams,
818 | ) -> GroupResponse:
819 | """
820 | Remove members from a group in ServiceNow.
821 |
822 | Args:
823 | config: Server configuration.
824 | auth_manager: Authentication manager.
825 | params: Parameters for removing members from the group.
826 |
827 | Returns:
828 | Response with the result of the operation.
829 | """
830 | success = True
831 | failed_members = []
832 |
833 | for member in params.members:
834 | # Get user ID if username is provided
835 | user_id = member
836 | if not member.startswith("sys_id:"):
837 | user = get_user(config, auth_manager, GetUserParams(user_name=member))
838 | if not user.get("success"):
839 | user = get_user(config, auth_manager, GetUserParams(email=member))
840 |
841 | if user.get("success"):
842 | user_id = user.get("user", {}).get("sys_id")
843 | else:
844 | success = False
845 | failed_members.append(member)
846 | continue
847 |
848 | # Find and delete the group membership
849 | api_url = f"{config.api_url}/table/sys_user_grmember"
850 | query_params = {
851 | "sysparm_query": f"group={params.group_id}^user={user_id}",
852 | "sysparm_limit": "1",
853 | }
854 |
855 | try:
856 | # First find the membership record
857 | response = requests.get(
858 | api_url,
859 | params=query_params,
860 | headers=auth_manager.get_headers(),
861 | timeout=config.timeout,
862 | )
863 | response.raise_for_status()
864 |
865 | result = response.json().get("result", [])
866 | if not result:
867 | success = False
868 | failed_members.append(member)
869 | continue
870 |
871 | # Then delete the membership record
872 | membership_id = result[0].get("sys_id")
873 | delete_url = f"{api_url}/{membership_id}"
874 |
875 | response = requests.delete(
876 | delete_url,
877 | headers=auth_manager.get_headers(),
878 | timeout=config.timeout,
879 | )
880 | response.raise_for_status()
881 |
882 | except requests.RequestException as e:
883 | logger.error(f"Failed to remove member '{member}' from group: {e}")
884 | success = False
885 | failed_members.append(member)
886 |
887 | if failed_members:
888 | message = f"Some members could not be removed from the group: {', '.join(failed_members)}"
889 | else:
890 | message = "All members removed from the group successfully"
891 |
892 | return GroupResponse(
893 | success=success,
894 | message=message,
895 | group_id=params.group_id,
896 | )
897 |
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/utils/tool_utils.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Any, Callable, Dict, Tuple, Type
2 |
3 | # Import all necessary tool implementation functions and params models
4 | # (This list needs to be kept complete and up-to-date)
5 | from servicenow_mcp.tools.catalog_optimization import (
6 | OptimizationRecommendationsParams,
7 | UpdateCatalogItemParams,
8 | )
9 | from servicenow_mcp.tools.catalog_optimization import (
10 | get_optimization_recommendations as get_optimization_recommendations_tool,
11 | )
12 | from servicenow_mcp.tools.catalog_optimization import (
13 | update_catalog_item as update_catalog_item_tool,
14 | )
15 | from servicenow_mcp.tools.catalog_tools import (
16 | CreateCatalogCategoryParams,
17 | GetCatalogItemParams,
18 | ListCatalogCategoriesParams,
19 | ListCatalogItemsParams,
20 | MoveCatalogItemsParams,
21 | UpdateCatalogCategoryParams,
22 | )
23 | from servicenow_mcp.tools.catalog_tools import (
24 | create_catalog_category as create_catalog_category_tool,
25 | )
26 | from servicenow_mcp.tools.catalog_tools import (
27 | get_catalog_item as get_catalog_item_tool,
28 | )
29 | from servicenow_mcp.tools.catalog_tools import (
30 | list_catalog_categories as list_catalog_categories_tool,
31 | )
32 | from servicenow_mcp.tools.catalog_tools import (
33 | list_catalog_items as list_catalog_items_tool,
34 | )
35 | from servicenow_mcp.tools.catalog_tools import (
36 | move_catalog_items as move_catalog_items_tool,
37 | )
38 | from servicenow_mcp.tools.catalog_tools import (
39 | update_catalog_category as update_catalog_category_tool,
40 | )
41 | from servicenow_mcp.tools.catalog_variables import (
42 | CreateCatalogItemVariableParams,
43 | ListCatalogItemVariablesParams,
44 | UpdateCatalogItemVariableParams,
45 | )
46 | from servicenow_mcp.tools.catalog_variables import (
47 | create_catalog_item_variable as create_catalog_item_variable_tool,
48 | )
49 | from servicenow_mcp.tools.catalog_variables import (
50 | list_catalog_item_variables as list_catalog_item_variables_tool,
51 | )
52 | from servicenow_mcp.tools.catalog_variables import (
53 | update_catalog_item_variable as update_catalog_item_variable_tool,
54 | )
55 | from servicenow_mcp.tools.change_tools import (
56 | AddChangeTaskParams,
57 | ApproveChangeParams,
58 | CreateChangeRequestParams,
59 | GetChangeRequestDetailsParams,
60 | ListChangeRequestsParams,
61 | RejectChangeParams,
62 | SubmitChangeForApprovalParams,
63 | UpdateChangeRequestParams,
64 | )
65 | from servicenow_mcp.tools.change_tools import (
66 | add_change_task as add_change_task_tool,
67 | )
68 | from servicenow_mcp.tools.change_tools import (
69 | approve_change as approve_change_tool,
70 | )
71 | from servicenow_mcp.tools.change_tools import (
72 | create_change_request as create_change_request_tool,
73 | )
74 | from servicenow_mcp.tools.change_tools import (
75 | get_change_request_details as get_change_request_details_tool,
76 | )
77 | from servicenow_mcp.tools.change_tools import (
78 | list_change_requests as list_change_requests_tool,
79 | )
80 | from servicenow_mcp.tools.change_tools import (
81 | reject_change as reject_change_tool,
82 | )
83 | from servicenow_mcp.tools.change_tools import (
84 | submit_change_for_approval as submit_change_for_approval_tool,
85 | )
86 | from servicenow_mcp.tools.change_tools import (
87 | update_change_request as update_change_request_tool,
88 | )
89 | from servicenow_mcp.tools.changeset_tools import (
90 | AddFileToChangesetParams,
91 | CommitChangesetParams,
92 | CreateChangesetParams,
93 | GetChangesetDetailsParams,
94 | ListChangesetsParams,
95 | PublishChangesetParams,
96 | UpdateChangesetParams,
97 | )
98 | from servicenow_mcp.tools.changeset_tools import (
99 | add_file_to_changeset as add_file_to_changeset_tool,
100 | )
101 | from servicenow_mcp.tools.changeset_tools import (
102 | commit_changeset as commit_changeset_tool,
103 | )
104 | from servicenow_mcp.tools.changeset_tools import (
105 | create_changeset as create_changeset_tool,
106 | )
107 | from servicenow_mcp.tools.changeset_tools import (
108 | get_changeset_details as get_changeset_details_tool,
109 | )
110 | from servicenow_mcp.tools.changeset_tools import (
111 | list_changesets as list_changesets_tool,
112 | )
113 | from servicenow_mcp.tools.changeset_tools import (
114 | publish_changeset as publish_changeset_tool,
115 | )
116 | from servicenow_mcp.tools.changeset_tools import (
117 | update_changeset as update_changeset_tool,
118 | )
119 | from servicenow_mcp.tools.incident_tools import (
120 | AddCommentParams,
121 | CreateIncidentParams,
122 | ListIncidentsParams,
123 | ResolveIncidentParams,
124 | UpdateIncidentParams,
125 | GetIncidentByNumberParams,
126 | )
127 | from servicenow_mcp.tools.incident_tools import (
128 | add_comment as add_comment_tool,
129 | )
130 | from servicenow_mcp.tools.incident_tools import (
131 | create_incident as create_incident_tool,
132 | )
133 | from servicenow_mcp.tools.incident_tools import (
134 | list_incidents as list_incidents_tool,
135 | )
136 | from servicenow_mcp.tools.incident_tools import (
137 | resolve_incident as resolve_incident_tool,
138 | )
139 | from servicenow_mcp.tools.incident_tools import (
140 | update_incident as update_incident_tool,
141 | )
142 | from servicenow_mcp.tools.incident_tools import (
143 | get_incident_by_number as get_incident_by_number_tool,
144 | )
145 | from servicenow_mcp.tools.knowledge_base import (
146 | CreateArticleParams,
147 | CreateKnowledgeBaseParams,
148 | GetArticleParams,
149 | ListArticlesParams,
150 | ListKnowledgeBasesParams,
151 | PublishArticleParams,
152 | UpdateArticleParams,
153 | )
154 | from servicenow_mcp.tools.knowledge_base import (
155 | CreateCategoryParams as CreateKBCategoryParams, # Aliased
156 | )
157 | from servicenow_mcp.tools.knowledge_base import (
158 | ListCategoriesParams as ListKBCategoriesParams, # Aliased
159 | )
160 | from servicenow_mcp.tools.knowledge_base import (
161 | create_article as create_article_tool,
162 | )
163 | from servicenow_mcp.tools.knowledge_base import (
164 | # create_category aliased in function call
165 | create_knowledge_base as create_knowledge_base_tool,
166 | )
167 | from servicenow_mcp.tools.knowledge_base import (
168 | get_article as get_article_tool,
169 | )
170 | from servicenow_mcp.tools.knowledge_base import (
171 | list_articles as list_articles_tool,
172 | )
173 | from servicenow_mcp.tools.knowledge_base import (
174 | # list_categories aliased in function call
175 | list_knowledge_bases as list_knowledge_bases_tool,
176 | )
177 | from servicenow_mcp.tools.knowledge_base import (
178 | publish_article as publish_article_tool,
179 | )
180 | from servicenow_mcp.tools.knowledge_base import (
181 | update_article as update_article_tool,
182 | )
183 | from servicenow_mcp.tools.script_include_tools import (
184 | CreateScriptIncludeParams,
185 | DeleteScriptIncludeParams,
186 | GetScriptIncludeParams,
187 | ListScriptIncludesParams,
188 | ScriptIncludeResponse,
189 | UpdateScriptIncludeParams,
190 | )
191 | from servicenow_mcp.tools.script_include_tools import (
192 | create_script_include as create_script_include_tool,
193 | )
194 | from servicenow_mcp.tools.script_include_tools import (
195 | delete_script_include as delete_script_include_tool,
196 | )
197 | from servicenow_mcp.tools.script_include_tools import (
198 | get_script_include as get_script_include_tool,
199 | )
200 | from servicenow_mcp.tools.script_include_tools import (
201 | list_script_includes as list_script_includes_tool,
202 | )
203 | from servicenow_mcp.tools.script_include_tools import (
204 | update_script_include as update_script_include_tool,
205 | )
206 | from servicenow_mcp.tools.user_tools import (
207 | AddGroupMembersParams,
208 | CreateGroupParams,
209 | CreateUserParams,
210 | GetUserParams,
211 | ListGroupsParams,
212 | ListUsersParams,
213 | RemoveGroupMembersParams,
214 | UpdateGroupParams,
215 | UpdateUserParams,
216 | )
217 | from servicenow_mcp.tools.user_tools import (
218 | add_group_members as add_group_members_tool,
219 | )
220 | from servicenow_mcp.tools.user_tools import (
221 | create_group as create_group_tool,
222 | )
223 | from servicenow_mcp.tools.user_tools import (
224 | create_user as create_user_tool,
225 | )
226 | from servicenow_mcp.tools.user_tools import (
227 | get_user as get_user_tool,
228 | )
229 | from servicenow_mcp.tools.user_tools import (
230 | list_groups as list_groups_tool,
231 | )
232 | from servicenow_mcp.tools.user_tools import (
233 | list_users as list_users_tool,
234 | )
235 | from servicenow_mcp.tools.user_tools import (
236 | remove_group_members as remove_group_members_tool,
237 | )
238 | from servicenow_mcp.tools.user_tools import (
239 | update_group as update_group_tool,
240 | )
241 | from servicenow_mcp.tools.user_tools import (
242 | update_user as update_user_tool,
243 | )
244 | from servicenow_mcp.tools.workflow_tools import (
245 | ActivateWorkflowParams,
246 | AddWorkflowActivityParams,
247 | CreateWorkflowParams,
248 | DeactivateWorkflowParams,
249 | DeleteWorkflowActivityParams,
250 | GetWorkflowActivitiesParams,
251 | GetWorkflowDetailsParams,
252 | ListWorkflowsParams,
253 | ListWorkflowVersionsParams,
254 | ReorderWorkflowActivitiesParams,
255 | UpdateWorkflowActivityParams,
256 | UpdateWorkflowParams,
257 | )
258 | from servicenow_mcp.tools.workflow_tools import (
259 | activate_workflow as activate_workflow_tool,
260 | )
261 | from servicenow_mcp.tools.workflow_tools import (
262 | add_workflow_activity as add_workflow_activity_tool,
263 | )
264 | from servicenow_mcp.tools.workflow_tools import (
265 | create_workflow as create_workflow_tool,
266 | )
267 | from servicenow_mcp.tools.workflow_tools import (
268 | deactivate_workflow as deactivate_workflow_tool,
269 | )
270 | from servicenow_mcp.tools.workflow_tools import (
271 | delete_workflow_activity as delete_workflow_activity_tool,
272 | )
273 | from servicenow_mcp.tools.workflow_tools import (
274 | get_workflow_activities as get_workflow_activities_tool,
275 | )
276 | from servicenow_mcp.tools.workflow_tools import (
277 | get_workflow_details as get_workflow_details_tool,
278 | )
279 | from servicenow_mcp.tools.workflow_tools import (
280 | list_workflow_versions as list_workflow_versions_tool,
281 | )
282 | from servicenow_mcp.tools.workflow_tools import (
283 | list_workflows as list_workflows_tool,
284 | )
285 | from servicenow_mcp.tools.workflow_tools import (
286 | reorder_workflow_activities as reorder_workflow_activities_tool,
287 | )
288 | from servicenow_mcp.tools.workflow_tools import (
289 | update_workflow as update_workflow_tool,
290 | )
291 | from servicenow_mcp.tools.workflow_tools import (
292 | update_workflow_activity as update_workflow_activity_tool,
293 | )
294 | from servicenow_mcp.tools.story_tools import (
295 | CreateStoryParams,
296 | UpdateStoryParams,
297 | ListStoriesParams,
298 | ListStoryDependenciesParams,
299 | CreateStoryDependencyParams,
300 | DeleteStoryDependencyParams,
301 | )
302 | from servicenow_mcp.tools.story_tools import (
303 | create_story as create_story_tool,
304 | update_story as update_story_tool,
305 | list_stories as list_stories_tool,
306 | list_story_dependencies as list_story_dependencies_tool,
307 | create_story_dependency as create_story_dependency_tool,
308 | delete_story_dependency as delete_story_dependency_tool,
309 | )
310 | from servicenow_mcp.tools.epic_tools import (
311 | CreateEpicParams,
312 | UpdateEpicParams,
313 | ListEpicsParams,
314 | )
315 | from servicenow_mcp.tools.epic_tools import (
316 | create_epic as create_epic_tool,
317 | update_epic as update_epic_tool,
318 | list_epics as list_epics_tool,
319 | )
320 | from servicenow_mcp.tools.scrum_task_tools import (
321 | CreateScrumTaskParams,
322 | UpdateScrumTaskParams,
323 | ListScrumTasksParams,
324 | )
325 | from servicenow_mcp.tools.scrum_task_tools import (
326 | create_scrum_task as create_scrum_task_tool,
327 | update_scrum_task as update_scrum_task_tool,
328 | list_scrum_tasks as list_scrum_tasks_tool,
329 | )
330 | from servicenow_mcp.tools.project_tools import (
331 | CreateProjectParams,
332 | UpdateProjectParams,
333 | ListProjectsParams,
334 | )
335 | from servicenow_mcp.tools.project_tools import (
336 | create_project as create_project_tool,
337 | update_project as update_project_tool,
338 | list_projects as list_projects_tool,
339 | )
340 |
341 | # Define a type alias for the Pydantic models or dataclasses used for params
342 | ParamsModel = Type[Any] # Use Type[Any] for broader compatibility initially
343 |
344 | # Define the structure of the tool definition tuple
345 | ToolDefinition = Tuple[
346 | Callable, # Implementation function
347 | ParamsModel, # Pydantic model for parameters
348 | Type, # Return type annotation (used for hints, not strictly enforced by low-level server)
349 | str, # Description
350 | str, # Serialization method ('str', 'json', 'dict', 'model_dump', etc.)
351 | ]
352 |
353 |
354 | def get_tool_definitions(
355 | create_kb_category_tool_impl: Callable, list_kb_categories_tool_impl: Callable
356 | ) -> Dict[str, ToolDefinition]:
357 | """
358 | Returns a dictionary containing definitions for all available ServiceNow tools.
359 |
360 | This centralizes the tool definitions for use in the server implementation.
361 | Pass aliased functions for KB categories directly.
362 |
363 | Returns:
364 | Dict[str, ToolDefinition]: A dictionary mapping tool names to their definitions.
365 | """
366 | tool_definitions: Dict[str, ToolDefinition] = {
367 | # Incident Tools
368 | "create_incident": (
369 | create_incident_tool,
370 | CreateIncidentParams,
371 | str,
372 | "Create a new incident in ServiceNow",
373 | "str",
374 | ),
375 | "update_incident": (
376 | update_incident_tool,
377 | UpdateIncidentParams,
378 | str,
379 | "Update an existing incident in ServiceNow",
380 | "str",
381 | ),
382 | "add_comment": (
383 | add_comment_tool,
384 | AddCommentParams,
385 | str,
386 | "Add a comment to an incident in ServiceNow",
387 | "str",
388 | ),
389 | "resolve_incident": (
390 | resolve_incident_tool,
391 | ResolveIncidentParams,
392 | str,
393 | "Resolve an incident in ServiceNow",
394 | "str",
395 | ),
396 | "list_incidents": (
397 | list_incidents_tool,
398 | ListIncidentsParams,
399 | str, # Expects JSON string
400 | "List incidents from ServiceNow",
401 | "json", # Tool returns list/dict, needs JSON dump
402 | ),
403 | "get_incident_by_number":(
404 | get_incident_by_number_tool,
405 | GetIncidentByNumberParams,
406 | str,
407 | "Incident details from ServiceNow",
408 | "json_dict"
409 | ),
410 | # Catalog Tools
411 | "list_catalog_items": (
412 | list_catalog_items_tool,
413 | ListCatalogItemsParams,
414 | str, # Expects JSON string
415 | "List service catalog items.",
416 | "json", # Tool returns list/dict
417 | ),
418 | "get_catalog_item": (
419 | get_catalog_item_tool,
420 | GetCatalogItemParams,
421 | str, # Expects JSON string
422 | "Get a specific service catalog item.",
423 | "json_dict", # Tool returns Pydantic model
424 | ),
425 | "list_catalog_categories": (
426 | list_catalog_categories_tool,
427 | ListCatalogCategoriesParams,
428 | str, # Expects JSON string
429 | "List service catalog categories.",
430 | "json", # Tool returns list/dict
431 | ),
432 | "create_catalog_category": (
433 | create_catalog_category_tool,
434 | CreateCatalogCategoryParams,
435 | str, # Expects JSON string
436 | "Create a new service catalog category.",
437 | "json_dict", # Tool returns Pydantic model
438 | ),
439 | "update_catalog_category": (
440 | update_catalog_category_tool,
441 | UpdateCatalogCategoryParams,
442 | str, # Expects JSON string
443 | "Update an existing service catalog category.",
444 | "json_dict", # Tool returns Pydantic model
445 | ),
446 | "move_catalog_items": (
447 | move_catalog_items_tool,
448 | MoveCatalogItemsParams,
449 | str, # Expects JSON string
450 | "Move catalog items to a different category.",
451 | "json_dict", # Tool returns Pydantic model
452 | ),
453 | "get_optimization_recommendations": (
454 | get_optimization_recommendations_tool,
455 | OptimizationRecommendationsParams,
456 | str, # Expects JSON string
457 | "Get optimization recommendations for the service catalog.",
458 | "json", # Tool returns list/dict
459 | ),
460 | "update_catalog_item": (
461 | update_catalog_item_tool,
462 | UpdateCatalogItemParams,
463 | str, # Expects JSON string
464 | "Update a service catalog item.",
465 | "json", # Tool returns Pydantic model
466 | ),
467 | # Catalog Variables
468 | "create_catalog_item_variable": (
469 | create_catalog_item_variable_tool,
470 | CreateCatalogItemVariableParams,
471 | Dict[str, Any], # Expects dict
472 | "Create a new catalog item variable",
473 | "dict", # Tool returns Pydantic model
474 | ),
475 | "list_catalog_item_variables": (
476 | list_catalog_item_variables_tool,
477 | ListCatalogItemVariablesParams,
478 | Dict[str, Any], # Expects dict
479 | "List catalog item variables",
480 | "dict", # Tool returns Pydantic model
481 | ),
482 | "update_catalog_item_variable": (
483 | update_catalog_item_variable_tool,
484 | UpdateCatalogItemVariableParams,
485 | Dict[str, Any], # Expects dict
486 | "Update a catalog item variable",
487 | "dict", # Tool returns Pydantic model
488 | ),
489 | # Change Management Tools
490 | "create_change_request": (
491 | create_change_request_tool,
492 | CreateChangeRequestParams,
493 | str,
494 | "Create a new change request in ServiceNow",
495 | "str",
496 | ),
497 | "update_change_request": (
498 | update_change_request_tool,
499 | UpdateChangeRequestParams,
500 | str,
501 | "Update an existing change request in ServiceNow",
502 | "str",
503 | ),
504 | "list_change_requests": (
505 | list_change_requests_tool,
506 | ListChangeRequestsParams,
507 | str, # Expects JSON string
508 | "List change requests from ServiceNow",
509 | "json", # Tool returns list/dict
510 | ),
511 | "get_change_request_details": (
512 | get_change_request_details_tool,
513 | GetChangeRequestDetailsParams,
514 | str, # Expects JSON string
515 | "Get detailed information about a specific change request",
516 | "json", # Tool returns list/dict
517 | ),
518 | "add_change_task": (
519 | add_change_task_tool,
520 | AddChangeTaskParams,
521 | str, # Expects JSON string
522 | "Add a task to a change request",
523 | "json_dict", # Tool returns Pydantic model
524 | ),
525 | "submit_change_for_approval": (
526 | submit_change_for_approval_tool,
527 | SubmitChangeForApprovalParams,
528 | str,
529 | "Submit a change request for approval",
530 | "str", # Tool returns simple message
531 | ),
532 | "approve_change": (
533 | approve_change_tool,
534 | ApproveChangeParams,
535 | str,
536 | "Approve a change request",
537 | "str", # Tool returns simple message
538 | ),
539 | "reject_change": (
540 | reject_change_tool,
541 | RejectChangeParams,
542 | str,
543 | "Reject a change request",
544 | "str", # Tool returns simple message
545 | ),
546 | # Workflow Management Tools
547 | "list_workflows": (
548 | list_workflows_tool,
549 | ListWorkflowsParams,
550 | str, # Expects JSON string
551 | "List workflows from ServiceNow",
552 | "json", # Tool returns list/dict
553 | ),
554 | "get_workflow_details": (
555 | get_workflow_details_tool,
556 | GetWorkflowDetailsParams,
557 | str, # Expects JSON string
558 | "Get detailed information about a specific workflow",
559 | "json", # Tool returns list/dict
560 | ),
561 | "list_workflow_versions": (
562 | list_workflow_versions_tool,
563 | ListWorkflowVersionsParams,
564 | str, # Expects JSON string
565 | "List workflow versions from ServiceNow",
566 | "json", # Tool returns list/dict
567 | ),
568 | "get_workflow_activities": (
569 | get_workflow_activities_tool,
570 | GetWorkflowActivitiesParams,
571 | str, # Expects JSON string
572 | "Get activities for a specific workflow",
573 | "json", # Tool returns list/dict
574 | ),
575 | "create_workflow": (
576 | create_workflow_tool,
577 | CreateWorkflowParams,
578 | str, # Expects JSON string
579 | "Create a new workflow in ServiceNow",
580 | "json_dict", # Tool returns Pydantic model
581 | ),
582 | "update_workflow": (
583 | update_workflow_tool,
584 | UpdateWorkflowParams,
585 | str, # Expects JSON string
586 | "Update an existing workflow in ServiceNow",
587 | "json_dict", # Tool returns Pydantic model
588 | ),
589 | "activate_workflow": (
590 | activate_workflow_tool,
591 | ActivateWorkflowParams,
592 | str,
593 | "Activate a workflow in ServiceNow",
594 | "str", # Tool returns simple message
595 | ),
596 | "deactivate_workflow": (
597 | deactivate_workflow_tool,
598 | DeactivateWorkflowParams,
599 | str,
600 | "Deactivate a workflow in ServiceNow",
601 | "str", # Tool returns simple message
602 | ),
603 | "add_workflow_activity": (
604 | add_workflow_activity_tool,
605 | AddWorkflowActivityParams,
606 | str, # Expects JSON string
607 | "Add a new activity to a workflow in ServiceNow",
608 | "json_dict", # Tool returns Pydantic model
609 | ),
610 | "update_workflow_activity": (
611 | update_workflow_activity_tool,
612 | UpdateWorkflowActivityParams,
613 | str, # Expects JSON string
614 | "Update an existing activity in a workflow",
615 | "json_dict", # Tool returns Pydantic model
616 | ),
617 | "delete_workflow_activity": (
618 | delete_workflow_activity_tool,
619 | DeleteWorkflowActivityParams,
620 | str,
621 | "Delete an activity from a workflow",
622 | "str", # Tool returns simple message
623 | ),
624 | "reorder_workflow_activities": (
625 | reorder_workflow_activities_tool,
626 | ReorderWorkflowActivitiesParams,
627 | str,
628 | "Reorder activities in a workflow",
629 | "str", # Tool returns simple message
630 | ),
631 | # Changeset Management Tools
632 | "list_changesets": (
633 | list_changesets_tool,
634 | ListChangesetsParams,
635 | str, # Expects JSON string
636 | "List changesets from ServiceNow",
637 | "json", # Tool returns list/dict
638 | ),
639 | "get_changeset_details": (
640 | get_changeset_details_tool,
641 | GetChangesetDetailsParams,
642 | str, # Expects JSON string
643 | "Get detailed information about a specific changeset",
644 | "json", # Tool returns list/dict
645 | ),
646 | "create_changeset": (
647 | create_changeset_tool,
648 | CreateChangesetParams,
649 | str, # Expects JSON string
650 | "Create a new changeset in ServiceNow",
651 | "json_dict", # Tool returns Pydantic model
652 | ),
653 | "update_changeset": (
654 | update_changeset_tool,
655 | UpdateChangesetParams,
656 | str, # Expects JSON string
657 | "Update an existing changeset in ServiceNow",
658 | "json_dict", # Tool returns Pydantic model
659 | ),
660 | "commit_changeset": (
661 | commit_changeset_tool,
662 | CommitChangesetParams,
663 | str,
664 | "Commit a changeset in ServiceNow",
665 | "str", # Tool returns simple message
666 | ),
667 | "publish_changeset": (
668 | publish_changeset_tool,
669 | PublishChangesetParams,
670 | str,
671 | "Publish a changeset in ServiceNow",
672 | "str", # Tool returns simple message
673 | ),
674 | "add_file_to_changeset": (
675 | add_file_to_changeset_tool,
676 | AddFileToChangesetParams,
677 | str,
678 | "Add a file to a changeset in ServiceNow",
679 | "str", # Tool returns simple message
680 | ),
681 | # Script Include Tools
682 | "list_script_includes": (
683 | list_script_includes_tool,
684 | ListScriptIncludesParams,
685 | Dict[str, Any], # Expects dict
686 | "List script includes from ServiceNow",
687 | "raw_dict", # Tool returns raw dict
688 | ),
689 | "get_script_include": (
690 | get_script_include_tool,
691 | GetScriptIncludeParams,
692 | Dict[str, Any], # Expects dict
693 | "Get a specific script include from ServiceNow",
694 | "raw_dict", # Tool returns raw dict
695 | ),
696 | "create_script_include": (
697 | create_script_include_tool,
698 | CreateScriptIncludeParams,
699 | ScriptIncludeResponse, # Expects Pydantic model
700 | "Create a new script include in ServiceNow",
701 | "raw_pydantic", # Tool returns Pydantic model
702 | ),
703 | "update_script_include": (
704 | update_script_include_tool,
705 | UpdateScriptIncludeParams,
706 | ScriptIncludeResponse, # Expects Pydantic model
707 | "Update an existing script include in ServiceNow",
708 | "raw_pydantic", # Tool returns Pydantic model
709 | ),
710 | "delete_script_include": (
711 | delete_script_include_tool,
712 | DeleteScriptIncludeParams,
713 | str, # Expects JSON string
714 | "Delete a script include in ServiceNow",
715 | "json_dict", # Tool returns Pydantic model
716 | ),
717 | # Knowledge Base Tools
718 | "create_knowledge_base": (
719 | create_knowledge_base_tool,
720 | CreateKnowledgeBaseParams,
721 | str, # Expects JSON string
722 | "Create a new knowledge base in ServiceNow",
723 | "json_dict", # Tool returns Pydantic model
724 | ),
725 | "list_knowledge_bases": (
726 | list_knowledge_bases_tool,
727 | ListKnowledgeBasesParams,
728 | Dict[str, Any], # Expects dict
729 | "List knowledge bases from ServiceNow",
730 | "raw_dict", # Tool returns raw dict
731 | ),
732 | # Use the passed-in implementations for aliased KB category tools
733 | "create_category": (
734 | create_kb_category_tool_impl, # Use passed function
735 | CreateKBCategoryParams,
736 | str, # Expects JSON string
737 | "Create a new category in a knowledge base",
738 | "json_dict", # Tool returns Pydantic model
739 | ),
740 | "create_article": (
741 | create_article_tool,
742 | CreateArticleParams,
743 | str, # Expects JSON string
744 | "Create a new knowledge article",
745 | "json_dict", # Tool returns Pydantic model
746 | ),
747 | "update_article": (
748 | update_article_tool,
749 | UpdateArticleParams,
750 | str, # Expects JSON string
751 | "Update an existing knowledge article",
752 | "json_dict", # Tool returns Pydantic model
753 | ),
754 | "publish_article": (
755 | publish_article_tool,
756 | PublishArticleParams,
757 | str, # Expects JSON string
758 | "Publish a knowledge article",
759 | "json_dict", # Tool returns Pydantic model
760 | ),
761 | "list_articles": (
762 | list_articles_tool,
763 | ListArticlesParams,
764 | Dict[str, Any], # Expects dict
765 | "List knowledge articles",
766 | "raw_dict", # Tool returns raw dict
767 | ),
768 | "get_article": (
769 | get_article_tool,
770 | GetArticleParams,
771 | Dict[str, Any], # Expects dict
772 | "Get a specific knowledge article by ID",
773 | "raw_dict", # Tool returns raw dict
774 | ),
775 | # Use the passed-in implementations for aliased KB category tools
776 | "list_categories": (
777 | list_kb_categories_tool_impl, # Use passed function
778 | ListKBCategoriesParams,
779 | Dict[str, Any], # Expects dict
780 | "List categories in a knowledge base",
781 | "raw_dict", # Tool returns raw dict
782 | ),
783 | # User Management Tools
784 | "create_user": (
785 | create_user_tool,
786 | CreateUserParams,
787 | Dict[str, Any], # Expects dict
788 | "Create a new user in ServiceNow",
789 | "raw_dict", # Tool returns raw dict
790 | ),
791 | "update_user": (
792 | update_user_tool,
793 | UpdateUserParams,
794 | Dict[str, Any], # Expects dict
795 | "Update an existing user in ServiceNow",
796 | "raw_dict",
797 | ),
798 | "get_user": (
799 | get_user_tool,
800 | GetUserParams,
801 | Dict[str, Any], # Expects dict
802 | "Get a specific user in ServiceNow",
803 | "raw_dict",
804 | ),
805 | "list_users": (
806 | list_users_tool,
807 | ListUsersParams,
808 | Dict[str, Any], # Expects dict
809 | "List users in ServiceNow",
810 | "raw_dict",
811 | ),
812 | "create_group": (
813 | create_group_tool,
814 | CreateGroupParams,
815 | Dict[str, Any], # Expects dict
816 | "Create a new group in ServiceNow",
817 | "raw_dict",
818 | ),
819 | "update_group": (
820 | update_group_tool,
821 | UpdateGroupParams,
822 | Dict[str, Any], # Expects dict
823 | "Update an existing group in ServiceNow",
824 | "raw_dict",
825 | ),
826 | "add_group_members": (
827 | add_group_members_tool,
828 | AddGroupMembersParams,
829 | Dict[str, Any], # Expects dict
830 | "Add members to an existing group in ServiceNow",
831 | "raw_dict",
832 | ),
833 | "remove_group_members": (
834 | remove_group_members_tool,
835 | RemoveGroupMembersParams,
836 | Dict[str, Any], # Expects dict
837 | "Remove members from an existing group in ServiceNow",
838 | "raw_dict",
839 | ),
840 | "list_groups": (
841 | list_groups_tool,
842 | ListGroupsParams,
843 | Dict[str, Any], # Expects dict
844 | "List groups from ServiceNow with optional filtering",
845 | "raw_dict",
846 | ),
847 | # Story Management Tools
848 | "create_story": (
849 | create_story_tool,
850 | CreateStoryParams,
851 | str,
852 | "Create a new story in ServiceNow",
853 | "str",
854 | ),
855 | "update_story": (
856 | update_story_tool,
857 | UpdateStoryParams,
858 | str,
859 | "Update an existing story in ServiceNow",
860 | "str",
861 | ),
862 | "list_stories": (
863 | list_stories_tool,
864 | ListStoriesParams,
865 | str, # Expects JSON string
866 | "List stories from ServiceNow",
867 | "json", # Tool returns list/dict
868 | ),
869 | "list_story_dependencies": (
870 | list_story_dependencies_tool,
871 | ListStoryDependenciesParams,
872 | str, # Expects JSON string
873 | "List story dependencies from ServiceNow",
874 | "json", # Tool returns list/dict
875 | ),
876 | "create_story_dependency": (
877 | create_story_dependency_tool,
878 | CreateStoryDependencyParams,
879 | str,
880 | "Create a dependency between two stories in ServiceNow",
881 | "str",
882 | ),
883 | "delete_story_dependency": (
884 | delete_story_dependency_tool,
885 | DeleteStoryDependencyParams,
886 | str,
887 | "Delete a story dependency in ServiceNow",
888 | "str",
889 | ),
890 | # Epic Management Tools
891 | "create_epic": (
892 | create_epic_tool,
893 | CreateEpicParams,
894 | str,
895 | "Create a new epic in ServiceNow",
896 | "str",
897 | ),
898 | "update_epic": (
899 | update_epic_tool,
900 | UpdateEpicParams,
901 | str,
902 | "Update an existing epic in ServiceNow",
903 | "str",
904 | ),
905 | "list_epics": (
906 | list_epics_tool,
907 | ListEpicsParams,
908 | str, # Expects JSON string
909 | "List epics from ServiceNow",
910 | "json", # Tool returns list/dict
911 | ),
912 | # Scrum Task Management Tools
913 | "create_scrum_task": (
914 | create_scrum_task_tool,
915 | CreateScrumTaskParams,
916 | str,
917 | "Create a new scrum task in ServiceNow",
918 | "str",
919 | ),
920 | "update_scrum_task": (
921 | update_scrum_task_tool,
922 | UpdateScrumTaskParams,
923 | str,
924 | "Update an existing scrum task in ServiceNow",
925 | "str",
926 | ),
927 | "list_scrum_tasks": (
928 | list_scrum_tasks_tool,
929 | ListScrumTasksParams,
930 | str, # Expects JSON string
931 | "List scrum tasks from ServiceNow",
932 | "json", # Tool returns list/dict
933 | ),
934 | # Project Management Tools
935 | "create_project": (
936 | create_project_tool,
937 | CreateProjectParams,
938 | str,
939 | "Create a new project in ServiceNow",
940 | "str",
941 | ),
942 | "update_project": (
943 | update_project_tool,
944 | UpdateProjectParams,
945 | str,
946 | "Update an existing project in ServiceNow",
947 | "str",
948 | ),
949 | "list_projects": (
950 | list_projects_tool,
951 | ListProjectsParams,
952 | str, # Expects JSON string
953 | "List projects from ServiceNow",
954 | "json", # Tool returns list/dict
955 | ),
956 | }
957 | return tool_definitions
958 |
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/change_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Change management tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing change requests 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 |
23 | class CreateChangeRequestParams(BaseModel):
24 | """Parameters for creating a change request."""
25 |
26 | short_description: str = Field(..., description="Short description of the change request")
27 | description: Optional[str] = Field(None, description="Detailed description of the change request")
28 | type: str = Field(..., description="Type of change (normal, standard, emergency)")
29 | risk: Optional[str] = Field(None, description="Risk level of the change")
30 | impact: Optional[str] = Field(None, description="Impact of the change")
31 | category: Optional[str] = Field(None, description="Category of the change")
32 | requested_by: Optional[str] = Field(None, description="User who requested the change")
33 | assignment_group: Optional[str] = Field(None, description="Group assigned to the change")
34 | start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
35 | end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
36 |
37 |
38 | class UpdateChangeRequestParams(BaseModel):
39 | """Parameters for updating a change request."""
40 |
41 | change_id: str = Field(..., description="Change request ID or sys_id")
42 | short_description: Optional[str] = Field(None, description="Short description of the change request")
43 | description: Optional[str] = Field(None, description="Detailed description of the change request")
44 | state: Optional[str] = Field(None, description="State of the change request")
45 | risk: Optional[str] = Field(None, description="Risk level of the change")
46 | impact: Optional[str] = Field(None, description="Impact of the change")
47 | category: Optional[str] = Field(None, description="Category of the change")
48 | assignment_group: Optional[str] = Field(None, description="Group assigned to the change")
49 | start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
50 | end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
51 | work_notes: Optional[str] = Field(None, description="Work notes to add to the change request")
52 |
53 |
54 | class ListChangeRequestsParams(BaseModel):
55 | """Parameters for listing change requests."""
56 |
57 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
58 | offset: Optional[int] = Field(0, description="Offset to start from")
59 | state: Optional[str] = Field(None, description="Filter by state")
60 | type: Optional[str] = Field(None, description="Filter by type (normal, standard, emergency)")
61 | category: Optional[str] = Field(None, description="Filter by category")
62 | assignment_group: Optional[str] = Field(None, description="Filter by assignment group")
63 | timeframe: Optional[str] = Field(None, description="Filter by timeframe (upcoming, in-progress, completed)")
64 | query: Optional[str] = Field(None, description="Additional query string")
65 |
66 |
67 | class GetChangeRequestDetailsParams(BaseModel):
68 | """Parameters for getting change request details."""
69 |
70 | change_id: str = Field(..., description="Change request ID or sys_id")
71 |
72 |
73 | class AddChangeTaskParams(BaseModel):
74 | """Parameters for adding a task to a change request."""
75 |
76 | change_id: str = Field(..., description="Change request ID or sys_id")
77 | short_description: str = Field(..., description="Short description of the task")
78 | description: Optional[str] = Field(None, description="Detailed description of the task")
79 | assigned_to: Optional[str] = Field(None, description="User assigned to the task")
80 | planned_start_date: Optional[str] = Field(None, description="Planned start date (YYYY-MM-DD HH:MM:SS)")
81 | planned_end_date: Optional[str] = Field(None, description="Planned end date (YYYY-MM-DD HH:MM:SS)")
82 |
83 |
84 | class SubmitChangeForApprovalParams(BaseModel):
85 | """Parameters for submitting a change request for approval."""
86 |
87 | change_id: str = Field(..., description="Change request ID or sys_id")
88 | approval_comments: Optional[str] = Field(None, description="Comments for the approval request")
89 |
90 |
91 | class ApproveChangeParams(BaseModel):
92 | """Parameters for approving a change request."""
93 |
94 | change_id: str = Field(..., description="Change request ID or sys_id")
95 | approver_id: Optional[str] = Field(None, description="ID of the approver")
96 | approval_comments: Optional[str] = Field(None, description="Comments for the approval")
97 |
98 |
99 | class RejectChangeParams(BaseModel):
100 | """Parameters for rejecting a change request."""
101 |
102 | change_id: str = Field(..., description="Change request ID or sys_id")
103 | approver_id: Optional[str] = Field(None, description="ID of the approver")
104 | rejection_reason: str = Field(..., description="Reason for rejection")
105 |
106 |
107 | def _unwrap_and_validate_params(params: Any, model_class: Type[T], required_fields: List[str] = None) -> Dict[str, Any]:
108 | """
109 | Helper function to unwrap and validate parameters.
110 |
111 | Args:
112 | params: The parameters to unwrap and validate.
113 | model_class: The Pydantic model class to validate against.
114 | required_fields: List of required field names.
115 |
116 | Returns:
117 | A tuple of (success, result) where result is either the validated parameters or an error message.
118 | """
119 | # Handle case where params might be wrapped in another dictionary
120 | if isinstance(params, dict) and len(params) == 1 and "params" in params and isinstance(params["params"], dict):
121 | logger.warning("Detected params wrapped in a 'params' key. Unwrapping...")
122 | params = params["params"]
123 |
124 | # Handle case where params might be a Pydantic model object
125 | if not isinstance(params, dict):
126 | try:
127 | # Try to convert to dict if it's a Pydantic model
128 | logger.warning("Params is not a dictionary. Attempting to convert...")
129 | params = params.dict() if hasattr(params, "dict") else dict(params)
130 | except Exception as e:
131 | logger.error(f"Failed to convert params to dictionary: {e}")
132 | return {
133 | "success": False,
134 | "message": f"Invalid parameters format. Expected a dictionary, got {type(params).__name__}",
135 | }
136 |
137 | # Validate required parameters are present
138 | if required_fields:
139 | for field in required_fields:
140 | if field not in params:
141 | return {
142 | "success": False,
143 | "message": f"Missing required parameter '{field}'",
144 | }
145 |
146 | try:
147 | # Validate parameters against the model
148 | validated_params = model_class(**params)
149 | return {
150 | "success": True,
151 | "params": validated_params,
152 | }
153 | except Exception as e:
154 | logger.error(f"Error validating parameters: {e}")
155 | return {
156 | "success": False,
157 | "message": f"Error validating parameters: {str(e)}",
158 | }
159 |
160 |
161 | def _get_instance_url(auth_manager: AuthManager, server_config: ServerConfig) -> Optional[str]:
162 | """
163 | Helper function to get the instance URL from either server_config or auth_manager.
164 |
165 | Args:
166 | auth_manager: The authentication manager.
167 | server_config: The server configuration.
168 |
169 | Returns:
170 | The instance URL if found, None otherwise.
171 | """
172 | if hasattr(server_config, 'instance_url'):
173 | return server_config.instance_url
174 | elif hasattr(auth_manager, 'instance_url'):
175 | return auth_manager.instance_url
176 | else:
177 | logger.error("Cannot find instance_url in either server_config or auth_manager")
178 | return None
179 |
180 |
181 | def _get_headers(auth_manager: Any, server_config: Any) -> Optional[Dict[str, str]]:
182 | """
183 | Helper function to get headers from either auth_manager or server_config.
184 |
185 | Args:
186 | auth_manager: The authentication manager or object passed as auth_manager.
187 | server_config: The server configuration or object passed as server_config.
188 |
189 | Returns:
190 | The headers if found, None otherwise.
191 | """
192 | # Try to get headers from auth_manager
193 | if hasattr(auth_manager, 'get_headers'):
194 | return auth_manager.get_headers()
195 |
196 | # If auth_manager doesn't have get_headers, try server_config
197 | if hasattr(server_config, 'get_headers'):
198 | return server_config.get_headers()
199 |
200 | # If neither has get_headers, check if auth_manager is actually a ServerConfig
201 | # and server_config is actually an AuthManager (parameters swapped)
202 | if hasattr(server_config, 'get_headers') and not hasattr(auth_manager, 'get_headers'):
203 | return server_config.get_headers()
204 |
205 | logger.error("Cannot find get_headers method in either auth_manager or server_config")
206 | return None
207 |
208 |
209 | def create_change_request(
210 | auth_manager: AuthManager,
211 | server_config: ServerConfig,
212 | params: Dict[str, Any],
213 | ) -> Dict[str, Any]:
214 | """
215 | Create a new change request in ServiceNow.
216 |
217 | Args:
218 | auth_manager: The authentication manager.
219 | server_config: The server configuration.
220 | params: The parameters for creating the change request.
221 |
222 | Returns:
223 | The created change request.
224 | """
225 | # Unwrap and validate parameters
226 | result = _unwrap_and_validate_params(
227 | params,
228 | CreateChangeRequestParams,
229 | required_fields=["short_description", "type"]
230 | )
231 |
232 | if not result["success"]:
233 | return result
234 |
235 | validated_params = result["params"]
236 |
237 | # Prepare the request data
238 | data = {
239 | "short_description": validated_params.short_description,
240 | "type": validated_params.type,
241 | }
242 |
243 | # Add optional fields if provided
244 | if validated_params.description:
245 | data["description"] = validated_params.description
246 | if validated_params.risk:
247 | data["risk"] = validated_params.risk
248 | if validated_params.impact:
249 | data["impact"] = validated_params.impact
250 | if validated_params.category:
251 | data["category"] = validated_params.category
252 | if validated_params.requested_by:
253 | data["requested_by"] = validated_params.requested_by
254 | if validated_params.assignment_group:
255 | data["assignment_group"] = validated_params.assignment_group
256 | if validated_params.start_date:
257 | data["start_date"] = validated_params.start_date
258 | if validated_params.end_date:
259 | data["end_date"] = validated_params.end_date
260 |
261 | # Get the instance URL
262 | instance_url = _get_instance_url(auth_manager, server_config)
263 | if not instance_url:
264 | return {
265 | "success": False,
266 | "message": "Cannot find instance_url in either server_config or auth_manager",
267 | }
268 |
269 | # Get the headers
270 | headers = _get_headers(auth_manager, server_config)
271 | if not headers:
272 | return {
273 | "success": False,
274 | "message": "Cannot find get_headers method in either auth_manager or server_config",
275 | }
276 |
277 | # Add Content-Type header
278 | headers["Content-Type"] = "application/json"
279 |
280 | # Make the API request
281 | url = f"{instance_url}/api/now/table/change_request"
282 |
283 | try:
284 | response = requests.post(url, json=data, headers=headers)
285 | response.raise_for_status()
286 |
287 | result = response.json()
288 |
289 | return {
290 | "success": True,
291 | "message": "Change request created successfully",
292 | "change_request": result["result"],
293 | }
294 | except requests.exceptions.RequestException as e:
295 | logger.error(f"Error creating change request: {e}")
296 | return {
297 | "success": False,
298 | "message": f"Error creating change request: {str(e)}",
299 | }
300 |
301 |
302 | def update_change_request(
303 | auth_manager: AuthManager,
304 | server_config: ServerConfig,
305 | params: Dict[str, Any],
306 | ) -> Dict[str, Any]:
307 | """
308 | Update an existing change request in ServiceNow.
309 |
310 | Args:
311 | auth_manager: The authentication manager.
312 | server_config: The server configuration.
313 | params: The parameters for updating the change request.
314 |
315 | Returns:
316 | The updated change request.
317 | """
318 | # Unwrap and validate parameters
319 | result = _unwrap_and_validate_params(
320 | params,
321 | UpdateChangeRequestParams,
322 | required_fields=["change_id"]
323 | )
324 |
325 | if not result["success"]:
326 | return result
327 |
328 | validated_params = result["params"]
329 |
330 | # Prepare the request data
331 | data = {}
332 |
333 | # Add fields if provided
334 | if validated_params.short_description:
335 | data["short_description"] = validated_params.short_description
336 | if validated_params.description:
337 | data["description"] = validated_params.description
338 | if validated_params.state:
339 | data["state"] = validated_params.state
340 | if validated_params.risk:
341 | data["risk"] = validated_params.risk
342 | if validated_params.impact:
343 | data["impact"] = validated_params.impact
344 | if validated_params.category:
345 | data["category"] = validated_params.category
346 | if validated_params.assignment_group:
347 | data["assignment_group"] = validated_params.assignment_group
348 | if validated_params.start_date:
349 | data["start_date"] = validated_params.start_date
350 | if validated_params.end_date:
351 | data["end_date"] = validated_params.end_date
352 | if validated_params.work_notes:
353 | data["work_notes"] = validated_params.work_notes
354 |
355 | # Get the instance URL
356 | instance_url = _get_instance_url(auth_manager, server_config)
357 | if not instance_url:
358 | return {
359 | "success": False,
360 | "message": "Cannot find instance_url in either server_config or auth_manager",
361 | }
362 |
363 | # Get the headers
364 | headers = _get_headers(auth_manager, server_config)
365 | if not headers:
366 | return {
367 | "success": False,
368 | "message": "Cannot find get_headers method in either auth_manager or server_config",
369 | }
370 |
371 | # Add Content-Type header
372 | headers["Content-Type"] = "application/json"
373 |
374 | # Make the API request
375 | url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
376 |
377 | try:
378 | response = requests.put(url, json=data, headers=headers)
379 | response.raise_for_status()
380 |
381 | result = response.json()
382 |
383 | return {
384 | "success": True,
385 | "message": "Change request updated successfully",
386 | "change_request": result["result"],
387 | }
388 | except requests.exceptions.RequestException as e:
389 | logger.error(f"Error updating change request: {e}")
390 | return {
391 | "success": False,
392 | "message": f"Error updating change request: {str(e)}",
393 | }
394 |
395 |
396 | def list_change_requests(
397 | auth_manager: AuthManager,
398 | server_config: ServerConfig,
399 | params: Dict[str, Any],
400 | ) -> Dict[str, Any]:
401 | """
402 | List change requests from ServiceNow.
403 |
404 | Args:
405 | auth_manager: The authentication manager.
406 | server_config: The server configuration.
407 | params: The parameters for listing change requests.
408 |
409 | Returns:
410 | A list of change requests.
411 | """
412 | # Unwrap and validate parameters
413 | result = _unwrap_and_validate_params(
414 | params,
415 | ListChangeRequestsParams
416 | )
417 |
418 | if not result["success"]:
419 | return result
420 |
421 | validated_params = result["params"]
422 |
423 | # Build the query
424 | query_parts = []
425 |
426 | if validated_params.state:
427 | query_parts.append(f"state={validated_params.state}")
428 | if validated_params.type:
429 | query_parts.append(f"type={validated_params.type}")
430 | if validated_params.category:
431 | query_parts.append(f"category={validated_params.category}")
432 | if validated_params.assignment_group:
433 | query_parts.append(f"assignment_group={validated_params.assignment_group}")
434 |
435 | # Handle timeframe filtering
436 | if validated_params.timeframe:
437 | now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
438 | if validated_params.timeframe == "upcoming":
439 | query_parts.append(f"start_date>{now}")
440 | elif validated_params.timeframe == "in-progress":
441 | query_parts.append(f"start_date<{now}^end_date>{now}")
442 | elif validated_params.timeframe == "completed":
443 | query_parts.append(f"end_date<{now}")
444 |
445 | # Add any additional query string
446 | if validated_params.query:
447 | query_parts.append(validated_params.query)
448 |
449 | # Combine query parts
450 | query = "^".join(query_parts) if query_parts else ""
451 |
452 | # Get the instance URL
453 | instance_url = _get_instance_url(auth_manager, server_config)
454 | if not instance_url:
455 | return {
456 | "success": False,
457 | "message": "Cannot find instance_url in either server_config or auth_manager",
458 | }
459 |
460 | # Get the headers
461 | headers = _get_headers(auth_manager, server_config)
462 | if not headers:
463 | return {
464 | "success": False,
465 | "message": "Cannot find get_headers method in either auth_manager or server_config",
466 | }
467 |
468 | # Make the API request
469 | url = f"{instance_url}/api/now/table/change_request"
470 |
471 | params = {
472 | "sysparm_limit": validated_params.limit,
473 | "sysparm_offset": validated_params.offset,
474 | "sysparm_query": query,
475 | "sysparm_display_value": "true",
476 | }
477 |
478 | try:
479 | response = requests.get(url, headers=headers, params=params)
480 | response.raise_for_status()
481 |
482 | result = response.json()
483 |
484 | # Handle the case where result["result"] is a list
485 | change_requests = result.get("result", [])
486 | count = len(change_requests)
487 |
488 | return {
489 | "success": True,
490 | "change_requests": change_requests,
491 | "count": count,
492 | "total": count, # Use count as total if total is not provided
493 | }
494 | except requests.exceptions.RequestException as e:
495 | logger.error(f"Error listing change requests: {e}")
496 | return {
497 | "success": False,
498 | "message": f"Error listing change requests: {str(e)}",
499 | }
500 |
501 |
502 | def get_change_request_details(
503 | auth_manager: AuthManager,
504 | server_config: ServerConfig,
505 | params: Dict[str, Any],
506 | ) -> Dict[str, Any]:
507 | """
508 | Get details of a change request from ServiceNow.
509 |
510 | Args:
511 | auth_manager: The authentication manager.
512 | server_config: The server configuration.
513 | params: The parameters for getting change request details.
514 |
515 | Returns:
516 | The change request details.
517 | """
518 | # Unwrap and validate parameters
519 | result = _unwrap_and_validate_params(
520 | params,
521 | GetChangeRequestDetailsParams,
522 | required_fields=["change_id"]
523 | )
524 |
525 | if not result["success"]:
526 | return result
527 |
528 | validated_params = result["params"]
529 |
530 | # Get the instance URL
531 | instance_url = _get_instance_url(auth_manager, server_config)
532 | if not instance_url:
533 | return {
534 | "success": False,
535 | "message": "Cannot find instance_url in either server_config or auth_manager",
536 | }
537 |
538 | # Get the headers
539 | headers = _get_headers(auth_manager, server_config)
540 | if not headers:
541 | return {
542 | "success": False,
543 | "message": "Cannot find get_headers method in either auth_manager or server_config",
544 | }
545 |
546 | # Make the API request
547 | url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
548 |
549 | params = {
550 | "sysparm_display_value": "true",
551 | }
552 |
553 | try:
554 | response = requests.get(url, headers=headers, params=params)
555 | response.raise_for_status()
556 |
557 | result = response.json()
558 |
559 | # Get tasks associated with this change request
560 | tasks_url = f"{instance_url}/api/now/table/change_task"
561 | tasks_params = {
562 | "sysparm_query": f"change_request={validated_params.change_id}",
563 | "sysparm_display_value": "true",
564 | }
565 |
566 | tasks_response = requests.get(tasks_url, headers=headers, params=tasks_params)
567 | tasks_response.raise_for_status()
568 |
569 | tasks_result = tasks_response.json()
570 |
571 | return {
572 | "success": True,
573 | "change_request": result["result"],
574 | "tasks": tasks_result["result"],
575 | }
576 | except requests.exceptions.RequestException as e:
577 | logger.error(f"Error getting change request details: {e}")
578 | return {
579 | "success": False,
580 | "message": f"Error getting change request details: {str(e)}",
581 | }
582 |
583 |
584 | def add_change_task(
585 | auth_manager: AuthManager,
586 | server_config: ServerConfig,
587 | params: Dict[str, Any],
588 | ) -> Dict[str, Any]:
589 | """
590 | Add a task to a change request in ServiceNow.
591 |
592 | Args:
593 | auth_manager: The authentication manager.
594 | server_config: The server configuration.
595 | params: The parameters for adding a change task.
596 |
597 | Returns:
598 | The created change task.
599 | """
600 | # Unwrap and validate parameters
601 | result = _unwrap_and_validate_params(
602 | params,
603 | AddChangeTaskParams,
604 | required_fields=["change_id", "short_description"]
605 | )
606 |
607 | if not result["success"]:
608 | return result
609 |
610 | validated_params = result["params"]
611 |
612 | # Prepare the request data
613 | data = {
614 | "change_request": validated_params.change_id,
615 | "short_description": validated_params.short_description,
616 | }
617 |
618 | # Add optional fields if provided
619 | if validated_params.description:
620 | data["description"] = validated_params.description
621 | if validated_params.assigned_to:
622 | data["assigned_to"] = validated_params.assigned_to
623 | if validated_params.planned_start_date:
624 | data["planned_start_date"] = validated_params.planned_start_date
625 | if validated_params.planned_end_date:
626 | data["planned_end_date"] = validated_params.planned_end_date
627 |
628 | # Get the instance URL
629 | instance_url = _get_instance_url(auth_manager, server_config)
630 | if not instance_url:
631 | return {
632 | "success": False,
633 | "message": "Cannot find instance_url in either server_config or auth_manager",
634 | }
635 |
636 | # Get the headers
637 | headers = _get_headers(auth_manager, server_config)
638 | if not headers:
639 | return {
640 | "success": False,
641 | "message": "Cannot find get_headers method in either auth_manager or server_config",
642 | }
643 |
644 | # Add Content-Type header
645 | headers["Content-Type"] = "application/json"
646 |
647 | # Make the API request
648 | url = f"{instance_url}/api/now/table/change_task"
649 |
650 | try:
651 | response = requests.post(url, json=data, headers=headers)
652 | response.raise_for_status()
653 |
654 | result = response.json()
655 |
656 | return {
657 | "success": True,
658 | "message": "Change task added successfully",
659 | "change_task": result["result"],
660 | }
661 | except requests.exceptions.RequestException as e:
662 | logger.error(f"Error adding change task: {e}")
663 | return {
664 | "success": False,
665 | "message": f"Error adding change task: {str(e)}",
666 | }
667 |
668 |
669 | def submit_change_for_approval(
670 | auth_manager: AuthManager,
671 | server_config: ServerConfig,
672 | params: Dict[str, Any],
673 | ) -> Dict[str, Any]:
674 | """
675 | Submit a change request for approval in ServiceNow.
676 |
677 | Args:
678 | auth_manager: The authentication manager.
679 | server_config: The server configuration.
680 | params: The parameters for submitting a change request for approval.
681 |
682 | Returns:
683 | The result of the submission.
684 | """
685 | # Unwrap and validate parameters
686 | result = _unwrap_and_validate_params(
687 | params,
688 | SubmitChangeForApprovalParams,
689 | required_fields=["change_id"]
690 | )
691 |
692 | if not result["success"]:
693 | return result
694 |
695 | validated_params = result["params"]
696 |
697 | # Prepare the request data
698 | data = {
699 | "state": "assess", # Set state to "assess" to submit for approval
700 | }
701 |
702 | # Add approval comments if provided
703 | if validated_params.approval_comments:
704 | data["work_notes"] = validated_params.approval_comments
705 |
706 | # Get the instance URL
707 | instance_url = _get_instance_url(auth_manager, server_config)
708 | if not instance_url:
709 | return {
710 | "success": False,
711 | "message": "Cannot find instance_url in either server_config or auth_manager",
712 | }
713 |
714 | # Get the headers
715 | headers = _get_headers(auth_manager, server_config)
716 | if not headers:
717 | return {
718 | "success": False,
719 | "message": "Cannot find get_headers method in either auth_manager or server_config",
720 | }
721 |
722 | # Add Content-Type header
723 | headers["Content-Type"] = "application/json"
724 |
725 | # Make the API request
726 | url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
727 |
728 | try:
729 | response = requests.patch(url, json=data, headers=headers)
730 | response.raise_for_status()
731 |
732 | # Now, create an approval request
733 | approval_url = f"{instance_url}/api/now/table/sysapproval_approver"
734 | approval_data = {
735 | "document_id": validated_params.change_id,
736 | "source_table": "change_request",
737 | "state": "requested",
738 | }
739 |
740 | approval_response = requests.post(approval_url, json=approval_data, headers=headers)
741 | approval_response.raise_for_status()
742 |
743 | approval_result = approval_response.json()
744 |
745 | return {
746 | "success": True,
747 | "message": "Change request submitted for approval successfully",
748 | "approval": approval_result["result"],
749 | }
750 | except requests.exceptions.RequestException as e:
751 | logger.error(f"Error submitting change for approval: {e}")
752 | return {
753 | "success": False,
754 | "message": f"Error submitting change for approval: {str(e)}",
755 | }
756 |
757 |
758 | def approve_change(
759 | auth_manager: AuthManager,
760 | server_config: ServerConfig,
761 | params: Dict[str, Any],
762 | ) -> Dict[str, Any]:
763 | """
764 | Approve a change request in ServiceNow.
765 |
766 | Args:
767 | auth_manager: The authentication manager.
768 | server_config: The server configuration.
769 | params: The parameters for approving a change request.
770 |
771 | Returns:
772 | The result of the approval.
773 | """
774 | # Unwrap and validate parameters
775 | result = _unwrap_and_validate_params(
776 | params,
777 | ApproveChangeParams,
778 | required_fields=["change_id"]
779 | )
780 |
781 | if not result["success"]:
782 | return result
783 |
784 | validated_params = result["params"]
785 |
786 | # Get the instance URL
787 | instance_url = _get_instance_url(auth_manager, server_config)
788 | if not instance_url:
789 | return {
790 | "success": False,
791 | "message": "Cannot find instance_url in either server_config or auth_manager",
792 | }
793 |
794 | # Get the headers
795 | headers = _get_headers(auth_manager, server_config)
796 | if not headers:
797 | return {
798 | "success": False,
799 | "message": "Cannot find get_headers method in either auth_manager or server_config",
800 | }
801 |
802 | # First, find the approval record
803 | approval_query_url = f"{instance_url}/api/now/table/sysapproval_approver"
804 |
805 | query_params = {
806 | "sysparm_query": f"document_id={validated_params.change_id}",
807 | "sysparm_limit": 1,
808 | }
809 |
810 | try:
811 | approval_response = requests.get(approval_query_url, headers=headers, params=query_params)
812 | approval_response.raise_for_status()
813 |
814 | approval_result = approval_response.json()
815 |
816 | if not approval_result.get("result") or len(approval_result["result"]) == 0:
817 | return {
818 | "success": False,
819 | "message": "No approval record found for this change request",
820 | }
821 |
822 | approval_id = approval_result["result"][0]["sys_id"]
823 |
824 | # Now, update the approval record to approved
825 | approval_update_url = f"{instance_url}/api/now/table/sysapproval_approver/{approval_id}"
826 | headers["Content-Type"] = "application/json"
827 |
828 | approval_data = {
829 | "state": "approved",
830 | }
831 |
832 | if validated_params.approval_comments:
833 | approval_data["comments"] = validated_params.approval_comments
834 |
835 | approval_update_response = requests.patch(approval_update_url, json=approval_data, headers=headers)
836 | approval_update_response.raise_for_status()
837 |
838 | # Finally, update the change request state to "implement"
839 | change_url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
840 |
841 | change_data = {
842 | "state": "implement", # This may vary depending on ServiceNow configuration
843 | }
844 |
845 | change_response = requests.patch(change_url, json=change_data, headers=headers)
846 | change_response.raise_for_status()
847 |
848 | return {
849 | "success": True,
850 | "message": "Change request approved successfully",
851 | }
852 | except requests.exceptions.RequestException as e:
853 | logger.error(f"Error approving change: {e}")
854 | return {
855 | "success": False,
856 | "message": f"Error approving change: {str(e)}",
857 | }
858 |
859 |
860 | def reject_change(
861 | auth_manager: AuthManager,
862 | server_config: ServerConfig,
863 | params: Dict[str, Any],
864 | ) -> Dict[str, Any]:
865 | """
866 | Reject a change request in ServiceNow.
867 |
868 | Args:
869 | auth_manager: The authentication manager.
870 | server_config: The server configuration.
871 | params: The parameters for rejecting a change request.
872 |
873 | Returns:
874 | The result of the rejection.
875 | """
876 | # Unwrap and validate parameters
877 | result = _unwrap_and_validate_params(
878 | params,
879 | RejectChangeParams,
880 | required_fields=["change_id", "rejection_reason"]
881 | )
882 |
883 | if not result["success"]:
884 | return result
885 |
886 | validated_params = result["params"]
887 |
888 | # Get the instance URL
889 | instance_url = _get_instance_url(auth_manager, server_config)
890 | if not instance_url:
891 | return {
892 | "success": False,
893 | "message": "Cannot find instance_url in either server_config or auth_manager",
894 | }
895 |
896 | # Get the headers
897 | headers = _get_headers(auth_manager, server_config)
898 | if not headers:
899 | return {
900 | "success": False,
901 | "message": "Cannot find get_headers method in either auth_manager or server_config",
902 | }
903 |
904 | # First, find the approval record
905 | approval_query_url = f"{instance_url}/api/now/table/sysapproval_approver"
906 |
907 | query_params = {
908 | "sysparm_query": f"document_id={validated_params.change_id}",
909 | "sysparm_limit": 1,
910 | }
911 |
912 | try:
913 | approval_response = requests.get(approval_query_url, headers=headers, params=query_params)
914 | approval_response.raise_for_status()
915 |
916 | approval_result = approval_response.json()
917 |
918 | if not approval_result.get("result") or len(approval_result["result"]) == 0:
919 | return {
920 | "success": False,
921 | "message": "No approval record found for this change request",
922 | }
923 |
924 | approval_id = approval_result["result"][0]["sys_id"]
925 |
926 | # Now, update the approval record to rejected
927 | approval_update_url = f"{instance_url}/api/now/table/sysapproval_approver/{approval_id}"
928 | headers["Content-Type"] = "application/json"
929 |
930 | approval_data = {
931 | "state": "rejected",
932 | "comments": validated_params.rejection_reason,
933 | }
934 |
935 | approval_update_response = requests.patch(approval_update_url, json=approval_data, headers=headers)
936 | approval_update_response.raise_for_status()
937 |
938 | # Finally, update the change request state to "canceled"
939 | change_url = f"{instance_url}/api/now/table/change_request/{validated_params.change_id}"
940 |
941 | change_data = {
942 | "state": "canceled", # This may vary depending on ServiceNow configuration
943 | "work_notes": f"Change request rejected: {validated_params.rejection_reason}",
944 | }
945 |
946 | change_response = requests.patch(change_url, json=change_data, headers=headers)
947 | change_response.raise_for_status()
948 |
949 | return {
950 | "success": True,
951 | "message": "Change request rejected successfully",
952 | }
953 | except requests.exceptions.RequestException as e:
954 | logger.error(f"Error rejecting change: {e}")
955 | return {
956 | "success": False,
957 | "message": f"Error rejecting change: {str(e)}",
958 | }
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/workflow_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Workflow management tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for viewing and managing workflows in ServiceNow.
5 | """
6 |
7 | import logging
8 | from datetime import datetime
9 | from typing import Any, Dict, List, Optional, Type, TypeVar, Union
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 |
23 | class ListWorkflowsParams(BaseModel):
24 | """Parameters for listing workflows."""
25 |
26 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
27 | offset: Optional[int] = Field(0, description="Offset to start from")
28 | active: Optional[bool] = Field(None, description="Filter by active status")
29 | name: Optional[str] = Field(None, description="Filter by name (contains)")
30 | query: Optional[str] = Field(None, description="Additional query string")
31 |
32 |
33 | class GetWorkflowDetailsParams(BaseModel):
34 | """Parameters for getting workflow details."""
35 |
36 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
37 | include_versions: Optional[bool] = Field(False, description="Include workflow versions")
38 |
39 |
40 | class ListWorkflowVersionsParams(BaseModel):
41 | """Parameters for listing workflow versions."""
42 |
43 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
44 | limit: Optional[int] = Field(10, description="Maximum number of records to return")
45 | offset: Optional[int] = Field(0, description="Offset to start from")
46 |
47 |
48 | class GetWorkflowActivitiesParams(BaseModel):
49 | """Parameters for getting workflow activities."""
50 |
51 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
52 | version: Optional[str] = Field(None, description="Specific version to get activities for")
53 |
54 |
55 | class CreateWorkflowParams(BaseModel):
56 | """Parameters for creating a new workflow."""
57 |
58 | name: str = Field(..., description="Name of the workflow")
59 | description: Optional[str] = Field(None, description="Description of the workflow")
60 | table: Optional[str] = Field(None, description="Table the workflow applies to")
61 | active: Optional[bool] = Field(True, description="Whether the workflow is active")
62 | attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the workflow")
63 |
64 |
65 | class UpdateWorkflowParams(BaseModel):
66 | """Parameters for updating a workflow."""
67 |
68 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
69 | name: Optional[str] = Field(None, description="Name of the workflow")
70 | description: Optional[str] = Field(None, description="Description of the workflow")
71 | table: Optional[str] = Field(None, description="Table the workflow applies to")
72 | active: Optional[bool] = Field(None, description="Whether the workflow is active")
73 | attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the workflow")
74 |
75 |
76 | class ActivateWorkflowParams(BaseModel):
77 | """Parameters for activating a workflow."""
78 |
79 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
80 |
81 |
82 | class DeactivateWorkflowParams(BaseModel):
83 | """Parameters for deactivating a workflow."""
84 |
85 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
86 |
87 |
88 | class AddWorkflowActivityParams(BaseModel):
89 | """Parameters for adding an activity to a workflow."""
90 |
91 | workflow_version_id: str = Field(..., description="Workflow version ID")
92 | name: str = Field(..., description="Name of the activity")
93 | description: Optional[str] = Field(None, description="Description of the activity")
94 | activity_type: str = Field(..., description="Type of activity (e.g., 'approval', 'task', 'notification')")
95 | attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the activity")
96 |
97 |
98 | class UpdateWorkflowActivityParams(BaseModel):
99 | """Parameters for updating a workflow activity."""
100 |
101 | activity_id: str = Field(..., description="Activity ID or sys_id")
102 | name: Optional[str] = Field(None, description="Name of the activity")
103 | description: Optional[str] = Field(None, description="Description of the activity")
104 | attributes: Optional[Dict[str, Any]] = Field(None, description="Additional attributes for the activity")
105 |
106 |
107 | class DeleteWorkflowActivityParams(BaseModel):
108 | """Parameters for deleting a workflow activity."""
109 |
110 | activity_id: str = Field(..., description="Activity ID or sys_id")
111 |
112 |
113 | class ReorderWorkflowActivitiesParams(BaseModel):
114 | """Parameters for reordering workflow activities."""
115 |
116 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
117 | activity_ids: List[str] = Field(..., description="List of activity IDs in the desired order")
118 |
119 |
120 | class DeleteWorkflowParams(BaseModel):
121 | """Parameters for deleting a workflow."""
122 |
123 | workflow_id: str = Field(..., description="Workflow ID or sys_id")
124 |
125 |
126 | def _unwrap_params(params: Any, param_class: Type[T]) -> Dict[str, Any]:
127 | """
128 | Unwrap parameters if they're wrapped in a Pydantic model.
129 | This helps handle cases where the parameters are passed as a model instead of a dict.
130 | """
131 | if isinstance(params, dict):
132 | return params
133 | if isinstance(params, param_class):
134 | return params.dict(exclude_none=True)
135 | return params
136 |
137 |
138 | def _get_auth_and_config(
139 | auth_manager_or_config: Union[AuthManager, ServerConfig],
140 | server_config_or_auth: Union[ServerConfig, AuthManager],
141 | ) -> tuple[AuthManager, ServerConfig]:
142 | """
143 | Get the correct auth_manager and server_config objects.
144 |
145 | This function handles the case where the parameters might be swapped.
146 |
147 | Args:
148 | auth_manager_or_config: Either an AuthManager or a ServerConfig.
149 | server_config_or_auth: Either a ServerConfig or an AuthManager.
150 |
151 | Returns:
152 | tuple[AuthManager, ServerConfig]: The correct auth_manager and server_config.
153 |
154 | Raises:
155 | ValueError: If the parameters are not of the expected types.
156 | """
157 | # Check if the parameters are in the correct order
158 | if isinstance(auth_manager_or_config, AuthManager) and isinstance(server_config_or_auth, ServerConfig):
159 | return auth_manager_or_config, server_config_or_auth
160 |
161 | # Check if the parameters are swapped
162 | if isinstance(auth_manager_or_config, ServerConfig) and isinstance(server_config_or_auth, AuthManager):
163 | return server_config_or_auth, auth_manager_or_config
164 |
165 | # If we get here, at least one of the parameters is not of the expected type
166 | if hasattr(auth_manager_or_config, "get_headers"):
167 | auth_manager = auth_manager_or_config
168 | elif hasattr(server_config_or_auth, "get_headers"):
169 | auth_manager = server_config_or_auth
170 | else:
171 | raise ValueError("Cannot find get_headers method in either auth_manager or server_config")
172 |
173 | if hasattr(auth_manager_or_config, "instance_url"):
174 | server_config = auth_manager_or_config
175 | elif hasattr(server_config_or_auth, "instance_url"):
176 | server_config = server_config_or_auth
177 | else:
178 | raise ValueError("Cannot find instance_url attribute in either auth_manager or server_config")
179 |
180 | return auth_manager, server_config
181 |
182 |
183 | def list_workflows(
184 | auth_manager: AuthManager,
185 | server_config: ServerConfig,
186 | params: Dict[str, Any],
187 | ) -> Dict[str, Any]:
188 | """
189 | List workflows from ServiceNow.
190 |
191 | Args:
192 | auth_manager: Authentication manager
193 | server_config: Server configuration
194 | params: Parameters for listing workflows
195 |
196 | Returns:
197 | Dictionary containing the list of workflows
198 | """
199 | params = _unwrap_params(params, ListWorkflowsParams)
200 |
201 | # Get the correct auth_manager and server_config
202 | try:
203 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
204 | except ValueError as e:
205 | logger.error(f"Error getting auth and config: {e}")
206 | return {"error": str(e)}
207 |
208 | # Convert parameters to ServiceNow query format
209 | query_params = {
210 | "sysparm_limit": params.get("limit", 10),
211 | "sysparm_offset": params.get("offset", 0),
212 | }
213 |
214 | # Build query string
215 | query_parts = []
216 |
217 | if params.get("active") is not None:
218 | query_parts.append(f"active={str(params['active']).lower()}")
219 |
220 | if params.get("name"):
221 | query_parts.append(f"nameLIKE{params['name']}")
222 |
223 | if params.get("query"):
224 | query_parts.append(params["query"])
225 |
226 | if query_parts:
227 | query_params["sysparm_query"] = "^".join(query_parts)
228 |
229 | # Make the API request
230 | try:
231 | headers = auth_manager.get_headers()
232 | url = f"{server_config.instance_url}/api/now/table/wf_workflow"
233 |
234 | response = requests.get(url, headers=headers, params=query_params)
235 | response.raise_for_status()
236 |
237 | result = response.json()
238 | return {
239 | "workflows": result.get("result", []),
240 | "count": len(result.get("result", [])),
241 | "total": int(response.headers.get("X-Total-Count", 0)),
242 | }
243 | except requests.RequestException as e:
244 | logger.error(f"Error listing workflows: {e}")
245 | return {"error": str(e)}
246 | except Exception as e:
247 | logger.error(f"Unexpected error listing workflows: {e}")
248 | return {"error": str(e)}
249 |
250 |
251 | def get_workflow_details(
252 | auth_manager: AuthManager,
253 | server_config: ServerConfig,
254 | params: Dict[str, Any],
255 | ) -> Dict[str, Any]:
256 | """
257 | Get detailed information about a specific workflow.
258 |
259 | Args:
260 | auth_manager: Authentication manager
261 | server_config: Server configuration
262 | params: Parameters for getting workflow details
263 |
264 | Returns:
265 | Dictionary containing the workflow details
266 | """
267 | params = _unwrap_params(params, GetWorkflowDetailsParams)
268 |
269 | # Get the correct auth_manager and server_config
270 | try:
271 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
272 | except ValueError as e:
273 | logger.error(f"Error getting auth and config: {e}")
274 | return {"error": str(e)}
275 |
276 | workflow_id = params.get("workflow_id")
277 | if not workflow_id:
278 | return {"error": "Workflow ID is required"}
279 |
280 | # Make the API request
281 | try:
282 | headers = auth_manager.get_headers()
283 | url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
284 |
285 | response = requests.get(url, headers=headers)
286 | response.raise_for_status()
287 |
288 | result = response.json()
289 | return {
290 | "workflow": result.get("result", {}),
291 | }
292 | except requests.RequestException as e:
293 | logger.error(f"Error getting workflow details: {e}")
294 | return {"error": str(e)}
295 | except Exception as e:
296 | logger.error(f"Unexpected error getting workflow details: {e}")
297 | return {"error": str(e)}
298 |
299 |
300 | def list_workflow_versions(
301 | auth_manager: AuthManager,
302 | server_config: ServerConfig,
303 | params: Dict[str, Any],
304 | ) -> Dict[str, Any]:
305 | """
306 | List versions of a specific workflow.
307 |
308 | Args:
309 | auth_manager: Authentication manager
310 | server_config: Server configuration
311 | params: Parameters for listing workflow versions
312 |
313 | Returns:
314 | Dict[str, Any]: List of workflow versions
315 | """
316 | # Unwrap parameters if needed
317 | params = _unwrap_params(params, ListWorkflowVersionsParams)
318 |
319 | # Get the correct auth_manager and server_config
320 | try:
321 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
322 | except ValueError as e:
323 | logger.error(f"Error getting auth and config: {e}")
324 | return {"error": str(e)}
325 |
326 | workflow_id = params.get("workflow_id")
327 | if not workflow_id:
328 | return {"error": "Workflow ID is required"}
329 |
330 | # Convert parameters to ServiceNow query format
331 | query_params = {
332 | "sysparm_query": f"workflow={workflow_id}",
333 | "sysparm_limit": params.get("limit", 10),
334 | "sysparm_offset": params.get("offset", 0),
335 | }
336 |
337 | # Make the API request
338 | try:
339 | headers = auth_manager.get_headers()
340 | url = f"{server_config.instance_url}/api/now/table/wf_workflow_version"
341 |
342 | response = requests.get(url, headers=headers, params=query_params)
343 | response.raise_for_status()
344 |
345 | result = response.json()
346 | return {
347 | "versions": result.get("result", []),
348 | "count": len(result.get("result", [])),
349 | "total": int(response.headers.get("X-Total-Count", 0)),
350 | "workflow_id": workflow_id,
351 | }
352 | except requests.RequestException as e:
353 | logger.error(f"Error listing workflow versions: {e}")
354 | return {"error": str(e)}
355 | except Exception as e:
356 | logger.error(f"Unexpected error listing workflow versions: {e}")
357 | return {"error": str(e)}
358 |
359 |
360 | def get_workflow_activities(
361 | auth_manager: AuthManager,
362 | server_config: ServerConfig,
363 | params: Dict[str, Any],
364 | ) -> Dict[str, Any]:
365 | """
366 | Get activities for a specific workflow.
367 |
368 | Args:
369 | auth_manager: Authentication manager
370 | server_config: Server configuration
371 | params: Parameters for getting workflow activities
372 |
373 | Returns:
374 | Dict[str, Any]: List of workflow activities
375 | """
376 | # Unwrap parameters if needed
377 | params = _unwrap_params(params, GetWorkflowActivitiesParams)
378 |
379 | # Get the correct auth_manager and server_config
380 | try:
381 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
382 | except ValueError as e:
383 | logger.error(f"Error getting auth and config: {e}")
384 | return {"error": str(e)}
385 |
386 | workflow_id = params.get("workflow_id")
387 | if not workflow_id:
388 | return {"error": "Workflow ID is required"}
389 |
390 | version_id = params.get("version")
391 |
392 | # If no version specified, get the latest published version
393 | if not version_id:
394 | try:
395 | headers = auth_manager.get_headers()
396 | version_url = f"{server_config.instance_url}/api/now/table/wf_workflow_version"
397 | version_params = {
398 | "sysparm_query": f"workflow={workflow_id}^published=true",
399 | "sysparm_limit": 1,
400 | "sysparm_orderby": "version DESC",
401 | }
402 |
403 | version_response = requests.get(version_url, headers=headers, params=version_params)
404 | version_response.raise_for_status()
405 |
406 | version_result = version_response.json()
407 | versions = version_result.get("result", [])
408 |
409 | if not versions:
410 | return {
411 | "error": f"No published versions found for workflow {workflow_id}",
412 | "workflow_id": workflow_id,
413 | }
414 |
415 | version_id = versions[0]["sys_id"]
416 | except requests.RequestException as e:
417 | logger.error(f"Error getting workflow version: {e}")
418 | return {"error": str(e)}
419 | except Exception as e:
420 | logger.error(f"Unexpected error getting workflow version: {e}")
421 | return {"error": str(e)}
422 |
423 | # Get activities for the version
424 | try:
425 | headers = auth_manager.get_headers()
426 | activities_url = f"{server_config.instance_url}/api/now/table/wf_activity"
427 | activities_params = {
428 | "sysparm_query": f"workflow_version={version_id}",
429 | "sysparm_orderby": "order",
430 | }
431 |
432 | activities_response = requests.get(activities_url, headers=headers, params=activities_params)
433 | activities_response.raise_for_status()
434 |
435 | activities_result = activities_response.json()
436 | return {
437 | "activities": activities_result.get("result", []),
438 | "count": len(activities_result.get("result", [])),
439 | "workflow_id": workflow_id,
440 | "version_id": version_id,
441 | }
442 | except requests.RequestException as e:
443 | logger.error(f"Error getting workflow activities: {e}")
444 | return {"error": str(e)}
445 | except Exception as e:
446 | logger.error(f"Unexpected error getting workflow activities: {e}")
447 | return {"error": str(e)}
448 |
449 |
450 | def create_workflow(
451 | auth_manager: AuthManager,
452 | server_config: ServerConfig,
453 | params: Dict[str, Any],
454 | ) -> Dict[str, Any]:
455 | """
456 | Create a new workflow in ServiceNow.
457 |
458 | Args:
459 | auth_manager: Authentication manager
460 | server_config: Server configuration
461 | params: Parameters for creating a workflow
462 |
463 | Returns:
464 | Dict[str, Any]: Created workflow details
465 | """
466 | # Unwrap parameters if needed
467 | params = _unwrap_params(params, CreateWorkflowParams)
468 |
469 | # Get the correct auth_manager and server_config
470 | try:
471 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
472 | except ValueError as e:
473 | logger.error(f"Error getting auth and config: {e}")
474 | return {"error": str(e)}
475 |
476 | # Validate required parameters
477 | if not params.get("name"):
478 | return {"error": "Workflow name is required"}
479 |
480 | # Prepare data for the API request
481 | data = {
482 | "name": params["name"],
483 | }
484 |
485 | if params.get("description"):
486 | data["description"] = params["description"]
487 |
488 | if params.get("table"):
489 | data["table"] = params["table"]
490 |
491 | if params.get("active") is not None:
492 | data["active"] = str(params["active"]).lower()
493 |
494 | if params.get("attributes"):
495 | # Add any additional attributes
496 | data.update(params["attributes"])
497 |
498 | # Make the API request
499 | try:
500 | headers = auth_manager.get_headers()
501 | url = f"{server_config.instance_url}/api/now/table/wf_workflow"
502 |
503 | response = requests.post(url, headers=headers, json=data)
504 | response.raise_for_status()
505 |
506 | result = response.json()
507 | return {
508 | "workflow": result.get("result", {}),
509 | "message": "Workflow created successfully",
510 | }
511 | except requests.RequestException as e:
512 | logger.error(f"Error creating workflow: {e}")
513 | return {"error": str(e)}
514 | except Exception as e:
515 | logger.error(f"Unexpected error creating workflow: {e}")
516 | return {"error": str(e)}
517 |
518 |
519 | def update_workflow(
520 | auth_manager: AuthManager,
521 | server_config: ServerConfig,
522 | params: Dict[str, Any],
523 | ) -> Dict[str, Any]:
524 | """
525 | Update an existing workflow in ServiceNow.
526 |
527 | Args:
528 | auth_manager: Authentication manager
529 | server_config: Server configuration
530 | params: Parameters for updating a workflow
531 |
532 | Returns:
533 | Dict[str, Any]: Updated workflow details
534 | """
535 | # Unwrap parameters if needed
536 | params = _unwrap_params(params, UpdateWorkflowParams)
537 |
538 | # Get the correct auth_manager and server_config
539 | try:
540 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
541 | except ValueError as e:
542 | logger.error(f"Error getting auth and config: {e}")
543 | return {"error": str(e)}
544 |
545 | workflow_id = params.get("workflow_id")
546 | if not workflow_id:
547 | return {"error": "Workflow ID is required"}
548 |
549 | # Prepare data for the API request
550 | data = {}
551 |
552 | if params.get("name"):
553 | data["name"] = params["name"]
554 |
555 | if params.get("description") is not None:
556 | data["description"] = params["description"]
557 |
558 | if params.get("table"):
559 | data["table"] = params["table"]
560 |
561 | if params.get("active") is not None:
562 | data["active"] = str(params["active"]).lower()
563 |
564 | if params.get("attributes"):
565 | # Add any additional attributes
566 | data.update(params["attributes"])
567 |
568 | if not data:
569 | return {"error": "No update parameters provided"}
570 |
571 | # Make the API request
572 | try:
573 | headers = auth_manager.get_headers()
574 | url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
575 |
576 | response = requests.patch(url, headers=headers, json=data)
577 | response.raise_for_status()
578 |
579 | result = response.json()
580 | return {
581 | "workflow": result.get("result", {}),
582 | "message": "Workflow updated successfully",
583 | }
584 | except requests.RequestException as e:
585 | logger.error(f"Error updating workflow: {e}")
586 | return {"error": str(e)}
587 | except Exception as e:
588 | logger.error(f"Unexpected error updating workflow: {e}")
589 | return {"error": str(e)}
590 |
591 |
592 | def activate_workflow(
593 | auth_manager: AuthManager,
594 | server_config: ServerConfig,
595 | params: Dict[str, Any],
596 | ) -> Dict[str, Any]:
597 | """
598 | Activate a workflow in ServiceNow.
599 |
600 | Args:
601 | auth_manager: Authentication manager
602 | server_config: Server configuration
603 | params: Parameters for activating a workflow
604 |
605 | Returns:
606 | Dict[str, Any]: Activated workflow details
607 | """
608 | # Unwrap parameters if needed
609 | params = _unwrap_params(params, ActivateWorkflowParams)
610 |
611 | # Get the correct auth_manager and server_config
612 | try:
613 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
614 | except ValueError as e:
615 | logger.error(f"Error getting auth and config: {e}")
616 | return {"error": str(e)}
617 |
618 | workflow_id = params.get("workflow_id")
619 | if not workflow_id:
620 | return {"error": "Workflow ID is required"}
621 |
622 | # Prepare data for the API request
623 | data = {
624 | "active": "true",
625 | }
626 |
627 | # Make the API request
628 | try:
629 | headers = auth_manager.get_headers()
630 | url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
631 |
632 | response = requests.patch(url, headers=headers, json=data)
633 | response.raise_for_status()
634 |
635 | result = response.json()
636 | return {
637 | "workflow": result.get("result", {}),
638 | "message": "Workflow activated successfully",
639 | }
640 | except requests.RequestException as e:
641 | logger.error(f"Error activating workflow: {e}")
642 | return {"error": str(e)}
643 | except Exception as e:
644 | logger.error(f"Unexpected error activating workflow: {e}")
645 | return {"error": str(e)}
646 |
647 |
648 | def deactivate_workflow(
649 | auth_manager: AuthManager,
650 | server_config: ServerConfig,
651 | params: Dict[str, Any],
652 | ) -> Dict[str, Any]:
653 | """
654 | Deactivate a workflow in ServiceNow.
655 |
656 | Args:
657 | auth_manager: Authentication manager
658 | server_config: Server configuration
659 | params: Parameters for deactivating a workflow
660 |
661 | Returns:
662 | Dict[str, Any]: Deactivated workflow details
663 | """
664 | # Unwrap parameters if needed
665 | params = _unwrap_params(params, DeactivateWorkflowParams)
666 |
667 | # Get the correct auth_manager and server_config
668 | try:
669 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
670 | except ValueError as e:
671 | logger.error(f"Error getting auth and config: {e}")
672 | return {"error": str(e)}
673 |
674 | workflow_id = params.get("workflow_id")
675 | if not workflow_id:
676 | return {"error": "Workflow ID is required"}
677 |
678 | # Prepare data for the API request
679 | data = {
680 | "active": "false",
681 | }
682 |
683 | # Make the API request
684 | try:
685 | headers = auth_manager.get_headers()
686 | url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
687 |
688 | response = requests.patch(url, headers=headers, json=data)
689 | response.raise_for_status()
690 |
691 | result = response.json()
692 | return {
693 | "workflow": result.get("result", {}),
694 | "message": "Workflow deactivated successfully",
695 | }
696 | except requests.RequestException as e:
697 | logger.error(f"Error deactivating workflow: {e}")
698 | return {"error": str(e)}
699 | except Exception as e:
700 | logger.error(f"Unexpected error deactivating workflow: {e}")
701 | return {"error": str(e)}
702 |
703 |
704 | def add_workflow_activity(
705 | auth_manager: AuthManager,
706 | server_config: ServerConfig,
707 | params: Dict[str, Any],
708 | ) -> Dict[str, Any]:
709 | """
710 | Add a new activity to a workflow.
711 |
712 | Args:
713 | auth_manager: Authentication manager
714 | server_config: Server configuration
715 | params: Parameters for adding a workflow activity
716 |
717 | Returns:
718 | Dict[str, Any]: Added workflow activity details
719 | """
720 | # Unwrap parameters if needed
721 | params = _unwrap_params(params, AddWorkflowActivityParams)
722 |
723 | # Get the correct auth_manager and server_config
724 | try:
725 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
726 | except ValueError as e:
727 | logger.error(f"Error getting auth and config: {e}")
728 | return {"error": str(e)}
729 |
730 | # Validate required parameters
731 | workflow_version_id = params.get("workflow_version_id")
732 | if not workflow_version_id:
733 | return {"error": "Workflow version ID is required"}
734 |
735 | activity_name = params.get("name")
736 | if not activity_name:
737 | return {"error": "Activity name is required"}
738 |
739 | # Prepare data for the API request
740 | data = {
741 | "workflow_version": workflow_version_id,
742 | "name": activity_name,
743 | }
744 |
745 | if params.get("description"):
746 | data["description"] = params["description"]
747 |
748 | if params.get("activity_type"):
749 | data["activity_type"] = params["activity_type"]
750 |
751 | if params.get("attributes"):
752 | # Add any additional attributes
753 | data.update(params["attributes"])
754 |
755 | # Make the API request
756 | try:
757 | headers = auth_manager.get_headers()
758 | url = f"{server_config.instance_url}/api/now/table/wf_activity"
759 |
760 | response = requests.post(url, headers=headers, json=data)
761 | response.raise_for_status()
762 |
763 | result = response.json()
764 | return {
765 | "activity": result.get("result", {}),
766 | "message": "Workflow activity added successfully",
767 | }
768 | except requests.RequestException as e:
769 | logger.error(f"Error adding workflow activity: {e}")
770 | return {"error": str(e)}
771 | except Exception as e:
772 | logger.error(f"Unexpected error adding workflow activity: {e}")
773 | return {"error": str(e)}
774 |
775 |
776 | def update_workflow_activity(
777 | auth_manager: AuthManager,
778 | server_config: ServerConfig,
779 | params: Dict[str, Any],
780 | ) -> Dict[str, Any]:
781 | """
782 | Update an existing activity in a workflow.
783 |
784 | Args:
785 | auth_manager: Authentication manager
786 | server_config: Server configuration
787 | params: Parameters for updating a workflow activity
788 |
789 | Returns:
790 | Dict[str, Any]: Updated workflow activity details
791 | """
792 | # Unwrap parameters if needed
793 | params = _unwrap_params(params, UpdateWorkflowActivityParams)
794 |
795 | # Get the correct auth_manager and server_config
796 | try:
797 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
798 | except ValueError as e:
799 | logger.error(f"Error getting auth and config: {e}")
800 | return {"error": str(e)}
801 |
802 | activity_id = params.get("activity_id")
803 | if not activity_id:
804 | return {"error": "Activity ID is required"}
805 |
806 | # Prepare data for the API request
807 | data = {}
808 |
809 | if params.get("name"):
810 | data["name"] = params["name"]
811 |
812 | if params.get("description") is not None:
813 | data["description"] = params["description"]
814 |
815 | if params.get("attributes"):
816 | # Add any additional attributes
817 | data.update(params["attributes"])
818 |
819 | if not data:
820 | return {"error": "No update parameters provided"}
821 |
822 | # Make the API request
823 | try:
824 | headers = auth_manager.get_headers()
825 | url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
826 |
827 | response = requests.patch(url, headers=headers, json=data)
828 | response.raise_for_status()
829 |
830 | result = response.json()
831 | return {
832 | "activity": result.get("result", {}),
833 | "message": "Activity updated successfully",
834 | }
835 | except requests.RequestException as e:
836 | logger.error(f"Error updating workflow activity: {e}")
837 | return {"error": str(e)}
838 | except Exception as e:
839 | logger.error(f"Unexpected error updating workflow activity: {e}")
840 | return {"error": str(e)}
841 |
842 |
843 | def delete_workflow_activity(
844 | auth_manager: AuthManager,
845 | server_config: ServerConfig,
846 | params: Dict[str, Any],
847 | ) -> Dict[str, Any]:
848 | """
849 | Delete an activity from a workflow.
850 |
851 | Args:
852 | auth_manager: Authentication manager
853 | server_config: Server configuration
854 | params: Parameters for deleting a workflow activity
855 |
856 | Returns:
857 | Dict[str, Any]: Result of the deletion operation
858 | """
859 | # Unwrap parameters if needed
860 | params = _unwrap_params(params, DeleteWorkflowActivityParams)
861 |
862 | # Get the correct auth_manager and server_config
863 | try:
864 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
865 | except ValueError as e:
866 | logger.error(f"Error getting auth and config: {e}")
867 | return {"error": str(e)}
868 |
869 | activity_id = params.get("activity_id")
870 | if not activity_id:
871 | return {"error": "Activity ID is required"}
872 |
873 | # Make the API request
874 | try:
875 | headers = auth_manager.get_headers()
876 | url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
877 |
878 | response = requests.delete(url, headers=headers)
879 | response.raise_for_status()
880 |
881 | return {
882 | "message": "Activity deleted successfully",
883 | "activity_id": activity_id,
884 | }
885 | except requests.RequestException as e:
886 | logger.error(f"Error deleting workflow activity: {e}")
887 | return {"error": str(e)}
888 | except Exception as e:
889 | logger.error(f"Unexpected error deleting workflow activity: {e}")
890 | return {"error": str(e)}
891 |
892 |
893 | def reorder_workflow_activities(
894 | auth_manager: AuthManager,
895 | server_config: ServerConfig,
896 | params: Dict[str, Any],
897 | ) -> Dict[str, Any]:
898 | """
899 | Reorder activities in a workflow.
900 |
901 | Args:
902 | auth_manager: Authentication manager
903 | server_config: Server configuration
904 | params: Parameters for reordering workflow activities
905 |
906 | Returns:
907 | Dict[str, Any]: Result of the reordering operation
908 | """
909 | # Unwrap parameters if needed
910 | params = _unwrap_params(params, ReorderWorkflowActivitiesParams)
911 |
912 | # Get the correct auth_manager and server_config
913 | try:
914 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
915 | except ValueError as e:
916 | logger.error(f"Error getting auth and config: {e}")
917 | return {"error": str(e)}
918 |
919 | workflow_id = params.get("workflow_id")
920 | if not workflow_id:
921 | return {"error": "Workflow ID is required"}
922 |
923 | activity_ids = params.get("activity_ids")
924 | if not activity_ids:
925 | return {"error": "Activity IDs are required"}
926 |
927 | # Make the API requests to update the order of each activity
928 | try:
929 | headers = auth_manager.get_headers()
930 | results = []
931 |
932 | for i, activity_id in enumerate(activity_ids):
933 | # Calculate the new order value (100, 200, 300, etc.)
934 | new_order = (i + 1) * 100
935 |
936 | url = f"{server_config.instance_url}/api/now/table/wf_activity/{activity_id}"
937 | data = {"order": new_order}
938 |
939 | try:
940 | response = requests.patch(url, headers=headers, json=data)
941 | response.raise_for_status()
942 |
943 | results.append({
944 | "activity_id": activity_id,
945 | "new_order": new_order,
946 | "success": True,
947 | })
948 | except requests.RequestException as e:
949 | logger.error(f"Error updating activity order: {e}")
950 | results.append({
951 | "activity_id": activity_id,
952 | "error": str(e),
953 | "success": False,
954 | })
955 |
956 | return {
957 | "message": "Activities reordered",
958 | "workflow_id": workflow_id,
959 | "results": results,
960 | }
961 | except Exception as e:
962 | logger.error(f"Unexpected error reordering workflow activities: {e}")
963 | return {"error": str(e)}
964 |
965 |
966 | def delete_workflow(
967 | auth_manager: AuthManager,
968 | server_config: ServerConfig,
969 | params: Dict[str, Any],
970 | ) -> Dict[str, Any]:
971 | """
972 | Delete a workflow from ServiceNow.
973 |
974 | Args:
975 | auth_manager: Authentication manager
976 | server_config: Server configuration
977 | params: Parameters for deleting a workflow
978 |
979 | Returns:
980 | Dict[str, Any]: Result of the deletion operation
981 | """
982 | # Unwrap parameters if needed
983 | params = _unwrap_params(params, DeleteWorkflowParams)
984 |
985 | # Get the correct auth_manager and server_config
986 | try:
987 | auth_manager, server_config = _get_auth_and_config(auth_manager, server_config)
988 | except ValueError as e:
989 | logger.error(f"Error getting auth and config: {e}")
990 | return {"error": str(e)}
991 |
992 | workflow_id = params.get("workflow_id")
993 | if not workflow_id:
994 | return {"error": "Workflow ID is required"}
995 |
996 | # Make the API request
997 | try:
998 | headers = auth_manager.get_headers()
999 | url = f"{server_config.instance_url}/api/now/table/wf_workflow/{workflow_id}"
1000 |
1001 | response = requests.delete(url, headers=headers)
1002 | response.raise_for_status()
1003 |
1004 | return {
1005 | "message": f"Workflow {workflow_id} deleted successfully",
1006 | "workflow_id": workflow_id,
1007 | }
1008 | except requests.RequestException as e:
1009 | logger.error(f"Error deleting workflow: {e}")
1010 | return {"error": str(e)}
1011 | except Exception as e:
1012 | logger.error(f"Unexpected error deleting workflow: {e}")
1013 | return {"error": str(e)}
```
--------------------------------------------------------------------------------
/src/servicenow_mcp/tools/knowledge_base.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Knowledge base tools for the ServiceNow MCP server.
3 |
4 | This module provides tools for managing knowledge bases, categories, and articles in ServiceNow.
5 | """
6 |
7 | import logging
8 | from typing import Any, Dict, 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 CreateKnowledgeBaseParams(BaseModel):
20 | """Parameters for creating a knowledge base."""
21 |
22 | title: str = Field(..., description="Title of the knowledge base")
23 | description: Optional[str] = Field(None, description="Description of the knowledge base")
24 | owner: Optional[str] = Field(None, description="The specified admin user or group")
25 | managers: Optional[str] = Field(None, description="Users who can manage this knowledge base")
26 | publish_workflow: Optional[str] = Field("Knowledge - Instant Publish", description="Publication workflow")
27 | retire_workflow: Optional[str] = Field("Knowledge - Instant Retire", description="Retirement workflow")
28 |
29 |
30 | class ListKnowledgeBasesParams(BaseModel):
31 | """Parameters for listing knowledge bases."""
32 |
33 | limit: int = Field(10, description="Maximum number of knowledge bases to return")
34 | offset: int = Field(0, description="Offset for pagination")
35 | active: Optional[bool] = Field(None, description="Filter by active status")
36 | query: Optional[str] = Field(None, description="Search query for knowledge bases")
37 |
38 |
39 | class CreateCategoryParams(BaseModel):
40 | """Parameters for creating a category in a knowledge base."""
41 |
42 | title: str = Field(..., description="Title of the category")
43 | description: Optional[str] = Field(None, description="Description of the category")
44 | knowledge_base: str = Field(..., description="The knowledge base to create the category in")
45 | parent_category: Optional[str] = Field(None, description="Parent category (if creating a subcategory). Sys_id refering to the parent category or sys_id of the parent table.")
46 | parent_table: Optional[str] = Field(None, description="Parent table (if creating a subcategory). Sys_id refering to the table where the parent category is defined.")
47 | active: bool = Field(True, description="Whether the category is active")
48 |
49 |
50 | class CreateArticleParams(BaseModel):
51 | """Parameters for creating a knowledge article."""
52 |
53 | title: str = Field(..., description="Title of the article")
54 | text: str = Field(..., description="The main body text for the article. Field supports html formatting and wiki markup based on the article_type. HTML is the default.")
55 | short_description: str = Field(..., description="Short description of the article")
56 | knowledge_base: str = Field(..., description="The knowledge base to create the article in")
57 | category: str = Field(..., description="Category for the article")
58 | keywords: Optional[str] = Field(None, description="Keywords for search")
59 | article_type: Optional[str] = Field("html", description="The type of article. Options are 'text' or 'wiki'. text lets the text field support html formatting. wiki lets the text field support wiki markup.")
60 |
61 |
62 | class UpdateArticleParams(BaseModel):
63 | """Parameters for updating a knowledge article."""
64 |
65 | article_id: str = Field(..., description="ID of the article to update")
66 | title: Optional[str] = Field(None, description="Updated title of the article")
67 | text: Optional[str] = Field(None, description="Updated main body text for the article. Field supports html formatting and wiki markup based on the article_type. HTML is the default.")
68 | short_description: Optional[str] = Field(None, description="Updated short description")
69 | category: Optional[str] = Field(None, description="Updated category for the article")
70 | keywords: Optional[str] = Field(None, description="Updated keywords for search")
71 |
72 |
73 | class PublishArticleParams(BaseModel):
74 | """Parameters for publishing a knowledge article."""
75 |
76 | article_id: str = Field(..., description="ID of the article to publish")
77 | workflow_state: Optional[str] = Field("published", description="The workflow state to set")
78 | workflow_version: Optional[str] = Field(None, description="The workflow version to use")
79 |
80 |
81 | class ListArticlesParams(BaseModel):
82 | """Parameters for listing knowledge articles."""
83 |
84 | limit: int = Field(10, description="Maximum number of articles to return")
85 | offset: int = Field(0, description="Offset for pagination")
86 | knowledge_base: Optional[str] = Field(None, description="Filter by knowledge base")
87 | category: Optional[str] = Field(None, description="Filter by category")
88 | query: Optional[str] = Field(None, description="Search query for articles")
89 | workflow_state: Optional[str] = Field(None, description="Filter by workflow state")
90 |
91 |
92 | class GetArticleParams(BaseModel):
93 | """Parameters for getting a knowledge article."""
94 |
95 | article_id: str = Field(..., description="ID of the article to get")
96 |
97 |
98 | class KnowledgeBaseResponse(BaseModel):
99 | """Response from knowledge base operations."""
100 |
101 | success: bool = Field(..., description="Whether the operation was successful")
102 | message: str = Field(..., description="Message describing the result")
103 | kb_id: Optional[str] = Field(None, description="ID of the affected knowledge base")
104 | kb_name: Optional[str] = Field(None, description="Name of the affected knowledge base")
105 |
106 |
107 | class CategoryResponse(BaseModel):
108 | """Response from category operations."""
109 |
110 | success: bool = Field(..., description="Whether the operation was successful")
111 | message: str = Field(..., description="Message describing the result")
112 | category_id: Optional[str] = Field(None, description="ID of the affected category")
113 | category_name: Optional[str] = Field(None, description="Name of the affected category")
114 |
115 |
116 | class ArticleResponse(BaseModel):
117 | """Response from article operations."""
118 |
119 | success: bool = Field(..., description="Whether the operation was successful")
120 | message: str = Field(..., description="Message describing the result")
121 | article_id: Optional[str] = Field(None, description="ID of the affected article")
122 | article_title: Optional[str] = Field(None, description="Title of the affected article")
123 | workflow_state: Optional[str] = Field(None, description="Current workflow state of the article")
124 |
125 |
126 | class ListCategoriesParams(BaseModel):
127 | """Parameters for listing categories in a knowledge base."""
128 |
129 | knowledge_base: Optional[str] = Field(None, description="Filter by knowledge base ID")
130 | parent_category: Optional[str] = Field(None, description="Filter by parent category ID")
131 | limit: int = Field(10, description="Maximum number of categories to return")
132 | offset: int = Field(0, description="Offset for pagination")
133 | active: Optional[bool] = Field(None, description="Filter by active status")
134 | query: Optional[str] = Field(None, description="Search query for categories")
135 |
136 |
137 | def create_knowledge_base(
138 | config: ServerConfig,
139 | auth_manager: AuthManager,
140 | params: CreateKnowledgeBaseParams,
141 | ) -> KnowledgeBaseResponse:
142 | """
143 | Create a new knowledge base in ServiceNow.
144 |
145 | Args:
146 | config: Server configuration.
147 | auth_manager: Authentication manager.
148 | params: Parameters for creating the knowledge base.
149 |
150 | Returns:
151 | Response with the created knowledge base details.
152 | """
153 | api_url = f"{config.api_url}/table/kb_knowledge_base"
154 |
155 | # Build request data
156 | data = {
157 | "title": params.title,
158 | }
159 |
160 | if params.description:
161 | data["description"] = params.description
162 | if params.owner:
163 | data["owner"] = params.owner
164 | if params.managers:
165 | data["kb_managers"] = params.managers
166 | if params.publish_workflow:
167 | data["workflow_publish"] = params.publish_workflow
168 | if params.retire_workflow:
169 | data["workflow_retire"] = params.retire_workflow
170 |
171 | # Make request
172 | try:
173 | response = requests.post(
174 | api_url,
175 | json=data,
176 | headers=auth_manager.get_headers(),
177 | timeout=config.timeout,
178 | )
179 | response.raise_for_status()
180 |
181 | result = response.json().get("result", {})
182 |
183 | return KnowledgeBaseResponse(
184 | success=True,
185 | message="Knowledge base created successfully",
186 | kb_id=result.get("sys_id"),
187 | kb_name=result.get("title"),
188 | )
189 |
190 | except requests.RequestException as e:
191 | logger.error(f"Failed to create knowledge base: {e}")
192 | return KnowledgeBaseResponse(
193 | success=False,
194 | message=f"Failed to create knowledge base: {str(e)}",
195 | )
196 |
197 |
198 | def list_knowledge_bases(
199 | config: ServerConfig,
200 | auth_manager: AuthManager,
201 | params: ListKnowledgeBasesParams,
202 | ) -> Dict[str, Any]:
203 | """
204 | List knowledge bases with filtering options.
205 |
206 | Args:
207 | config: Server configuration.
208 | auth_manager: Authentication manager.
209 | params: Parameters for listing knowledge bases.
210 |
211 | Returns:
212 | Dictionary with list of knowledge bases and metadata.
213 | """
214 | api_url = f"{config.api_url}/table/kb_knowledge_base"
215 |
216 | # Build query parameters
217 | query_params = {
218 | "sysparm_limit": params.limit,
219 | "sysparm_offset": params.offset,
220 | "sysparm_display_value": "true",
221 | }
222 |
223 | # Build query string
224 | query_parts = []
225 | if params.active is not None:
226 | query_parts.append(f"active={str(params.active).lower()}")
227 | if params.query:
228 | query_parts.append(f"titleLIKE{params.query}^ORdescriptionLIKE{params.query}")
229 |
230 | if query_parts:
231 | query_params["sysparm_query"] = "^".join(query_parts)
232 |
233 | # Make request
234 | try:
235 | response = requests.get(
236 | api_url,
237 | params=query_params,
238 | headers=auth_manager.get_headers(),
239 | timeout=config.timeout,
240 | )
241 | response.raise_for_status()
242 |
243 | # Get the JSON response
244 | json_response = response.json()
245 |
246 | # Safely extract the result
247 | if isinstance(json_response, dict) and "result" in json_response:
248 | result = json_response.get("result", [])
249 | else:
250 | logger.error("Unexpected response format: %s", json_response)
251 | return {
252 | "success": False,
253 | "message": "Unexpected response format",
254 | "knowledge_bases": [],
255 | "count": 0,
256 | "limit": params.limit,
257 | "offset": params.offset,
258 | }
259 |
260 | # Transform the results - create a simpler structure
261 | knowledge_bases = []
262 |
263 | # Handle either string or list
264 | if isinstance(result, list):
265 | for kb_item in result:
266 | if not isinstance(kb_item, dict):
267 | logger.warning("Skipping non-dictionary KB item: %s", kb_item)
268 | continue
269 |
270 | # Safely extract values
271 | kb_id = kb_item.get("sys_id", "")
272 | title = kb_item.get("title", "")
273 | description = kb_item.get("description", "")
274 |
275 | # Extract nested values safely
276 | owner = ""
277 | if isinstance(kb_item.get("owner"), dict):
278 | owner = kb_item["owner"].get("display_value", "")
279 |
280 | managers = ""
281 | if isinstance(kb_item.get("kb_managers"), dict):
282 | managers = kb_item["kb_managers"].get("display_value", "")
283 |
284 | active = False
285 | if kb_item.get("active") == "true":
286 | active = True
287 |
288 | created = kb_item.get("sys_created_on", "")
289 | updated = kb_item.get("sys_updated_on", "")
290 |
291 | knowledge_bases.append({
292 | "id": kb_id,
293 | "title": title,
294 | "description": description,
295 | "owner": owner,
296 | "managers": managers,
297 | "active": active,
298 | "created": created,
299 | "updated": updated,
300 | })
301 | else:
302 | logger.warning("Result is not a list: %s", result)
303 |
304 | return {
305 | "success": True,
306 | "message": f"Found {len(knowledge_bases)} knowledge bases",
307 | "knowledge_bases": knowledge_bases,
308 | "count": len(knowledge_bases),
309 | "limit": params.limit,
310 | "offset": params.offset,
311 | }
312 |
313 | except requests.RequestException as e:
314 | logger.error(f"Failed to list knowledge bases: {e}")
315 | return {
316 | "success": False,
317 | "message": f"Failed to list knowledge bases: {str(e)}",
318 | "knowledge_bases": [],
319 | "count": 0,
320 | "limit": params.limit,
321 | "offset": params.offset,
322 | }
323 |
324 |
325 | def create_category(
326 | config: ServerConfig,
327 | auth_manager: AuthManager,
328 | params: CreateCategoryParams,
329 | ) -> CategoryResponse:
330 | """
331 | Create a new category in a knowledge base.
332 |
333 | Args:
334 | config: Server configuration.
335 | auth_manager: Authentication manager.
336 | params: Parameters for creating the category.
337 |
338 | Returns:
339 | Response with the created category details.
340 | """
341 | api_url = f"{config.api_url}/table/kb_category"
342 |
343 | # Build request data
344 | data = {
345 | "label": params.title,
346 | "kb_knowledge_base": params.knowledge_base,
347 | # Convert boolean to string "true"/"false" as ServiceNow expects
348 | "active": str(params.active).lower(),
349 | }
350 |
351 | if params.description:
352 | data["description"] = params.description
353 | if params.parent_category:
354 | data["parent"] = params.parent_category
355 | if params.parent_table:
356 | data["parent_table"] = params.parent_table
357 |
358 | # Log the request data for debugging
359 | logger.debug(f"Creating category with data: {data}")
360 |
361 | # Make request
362 | try:
363 | response = requests.post(
364 | api_url,
365 | json=data,
366 | headers=auth_manager.get_headers(),
367 | timeout=config.timeout,
368 | )
369 | response.raise_for_status()
370 |
371 | result = response.json().get("result", {})
372 | logger.debug(f"Category creation response: {result}")
373 |
374 | # Log the specific fields to check the knowledge base assignment
375 | if "kb_knowledge_base" in result:
376 | logger.debug(f"Knowledge base in response: {result['kb_knowledge_base']}")
377 |
378 | # Log the active status
379 | if "active" in result:
380 | logger.debug(f"Active status in response: {result['active']}")
381 |
382 | return CategoryResponse(
383 | success=True,
384 | message="Category created successfully",
385 | category_id=result.get("sys_id"),
386 | category_name=result.get("label"),
387 | )
388 |
389 | except requests.RequestException as e:
390 | logger.error(f"Failed to create category: {e}")
391 | return CategoryResponse(
392 | success=False,
393 | message=f"Failed to create category: {str(e)}",
394 | )
395 |
396 |
397 | def create_article(
398 | config: ServerConfig,
399 | auth_manager: AuthManager,
400 | params: CreateArticleParams,
401 | ) -> ArticleResponse:
402 | """
403 | Create a new knowledge article.
404 |
405 | Args:
406 | config: Server configuration.
407 | auth_manager: Authentication manager.
408 | params: Parameters for creating the article.
409 |
410 | Returns:
411 | Response with the created article details.
412 | """
413 | api_url = f"{config.api_url}/table/kb_knowledge"
414 |
415 | # Build request data
416 | data = {
417 | "short_description": params.short_description,
418 | "text": params.text,
419 | "kb_knowledge_base": params.knowledge_base,
420 | "kb_category": params.category,
421 | "article_type": params.article_type,
422 | }
423 |
424 | if params.title:
425 | data["short_description"] = params.title
426 | if params.keywords:
427 | data["keywords"] = params.keywords
428 |
429 | # Make request
430 | try:
431 | response = requests.post(
432 | api_url,
433 | json=data,
434 | headers=auth_manager.get_headers(),
435 | timeout=config.timeout,
436 | )
437 | response.raise_for_status()
438 |
439 | result = response.json().get("result", {})
440 |
441 | return ArticleResponse(
442 | success=True,
443 | message="Article created successfully",
444 | article_id=result.get("sys_id"),
445 | article_title=result.get("short_description"),
446 | workflow_state=result.get("workflow_state"),
447 | )
448 |
449 | except requests.RequestException as e:
450 | logger.error(f"Failed to create article: {e}")
451 | return ArticleResponse(
452 | success=False,
453 | message=f"Failed to create article: {str(e)}",
454 | )
455 |
456 |
457 | def update_article(
458 | config: ServerConfig,
459 | auth_manager: AuthManager,
460 | params: UpdateArticleParams,
461 | ) -> ArticleResponse:
462 | """
463 | Update an existing knowledge article.
464 |
465 | Args:
466 | config: Server configuration.
467 | auth_manager: Authentication manager.
468 | params: Parameters for updating the article.
469 |
470 | Returns:
471 | Response with the updated article details.
472 | """
473 | api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
474 |
475 | # Build request data
476 | data = {}
477 |
478 | if params.title:
479 | data["short_description"] = params.title
480 | if params.text:
481 | data["text"] = params.text
482 | if params.short_description:
483 | data["short_description"] = params.short_description
484 | if params.category:
485 | data["kb_category"] = params.category
486 | if params.keywords:
487 | data["keywords"] = params.keywords
488 |
489 | # Make request
490 | try:
491 | response = requests.patch(
492 | api_url,
493 | json=data,
494 | headers=auth_manager.get_headers(),
495 | timeout=config.timeout,
496 | )
497 | response.raise_for_status()
498 |
499 | result = response.json().get("result", {})
500 |
501 | return ArticleResponse(
502 | success=True,
503 | message="Article updated successfully",
504 | article_id=params.article_id,
505 | article_title=result.get("short_description"),
506 | workflow_state=result.get("workflow_state"),
507 | )
508 |
509 | except requests.RequestException as e:
510 | logger.error(f"Failed to update article: {e}")
511 | return ArticleResponse(
512 | success=False,
513 | message=f"Failed to update article: {str(e)}",
514 | )
515 |
516 |
517 | def publish_article(
518 | config: ServerConfig,
519 | auth_manager: AuthManager,
520 | params: PublishArticleParams,
521 | ) -> ArticleResponse:
522 | """
523 | Publish a knowledge article.
524 |
525 | Args:
526 | config: Server configuration.
527 | auth_manager: Authentication manager.
528 | params: Parameters for publishing the article.
529 |
530 | Returns:
531 | Response with the published article details.
532 | """
533 | api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
534 |
535 | # Build request data
536 | data = {
537 | "workflow_state": params.workflow_state,
538 | }
539 |
540 | if params.workflow_version:
541 | data["workflow_version"] = params.workflow_version
542 |
543 | # Make request
544 | try:
545 | response = requests.patch(
546 | api_url,
547 | json=data,
548 | headers=auth_manager.get_headers(),
549 | timeout=config.timeout,
550 | )
551 | response.raise_for_status()
552 |
553 | result = response.json().get("result", {})
554 |
555 | return ArticleResponse(
556 | success=True,
557 | message="Article published successfully",
558 | article_id=params.article_id,
559 | article_title=result.get("short_description"),
560 | workflow_state=result.get("workflow_state"),
561 | )
562 |
563 | except requests.RequestException as e:
564 | logger.error(f"Failed to publish article: {e}")
565 | return ArticleResponse(
566 | success=False,
567 | message=f"Failed to publish article: {str(e)}",
568 | )
569 |
570 |
571 | def list_articles(
572 | config: ServerConfig,
573 | auth_manager: AuthManager,
574 | params: ListArticlesParams,
575 | ) -> Dict[str, Any]:
576 | """
577 | List knowledge articles with filtering options.
578 |
579 | Args:
580 | config: Server configuration.
581 | auth_manager: Authentication manager.
582 | params: Parameters for listing articles.
583 |
584 | Returns:
585 | Dictionary with list of articles and metadata.
586 | """
587 | api_url = f"{config.api_url}/table/kb_knowledge"
588 |
589 | # Build query parameters
590 | query_params = {
591 | "sysparm_limit": params.limit,
592 | "sysparm_offset": params.offset,
593 | "sysparm_display_value": "all",
594 | }
595 |
596 | # Build query string
597 | query_parts = []
598 | if params.knowledge_base:
599 | query_parts.append(f"kb_knowledge_base.sys_id={params.knowledge_base}")
600 | if params.category:
601 | query_parts.append(f"kb_category.sys_id={params.category}")
602 | if params.workflow_state:
603 | query_parts.append(f"workflow_state={params.workflow_state}")
604 | if params.query:
605 | query_parts.append(f"short_descriptionLIKE{params.query}^ORtextLIKE{params.query}")
606 |
607 | if query_parts:
608 | query_string = "^".join(query_parts)
609 | logger.debug(f"Constructed article query string: {query_string}")
610 | query_params["sysparm_query"] = query_string
611 |
612 | # Log the query parameters for debugging
613 | logger.debug(f"Listing articles with query params: {query_params}")
614 |
615 | # Make request
616 | try:
617 | response = requests.get(
618 | api_url,
619 | params=query_params,
620 | headers=auth_manager.get_headers(),
621 | timeout=config.timeout,
622 | )
623 | response.raise_for_status()
624 |
625 | # Get the JSON response
626 | json_response = response.json()
627 | logger.debug(f"Article listing raw response: {json_response}")
628 |
629 | # Safely extract the result
630 | if isinstance(json_response, dict) and "result" in json_response:
631 | result = json_response.get("result", [])
632 | else:
633 | logger.error("Unexpected response format: %s", json_response)
634 | return {
635 | "success": False,
636 | "message": f"Unexpected response format",
637 | "articles": [],
638 | "count": 0,
639 | "limit": params.limit,
640 | "offset": params.offset,
641 | }
642 |
643 | # Transform the results
644 | articles = []
645 |
646 | # Handle either string or list
647 | if isinstance(result, list):
648 | for article_item in result:
649 | if not isinstance(article_item, dict):
650 | logger.warning("Skipping non-dictionary article item: %s", article_item)
651 | continue
652 |
653 | # Safely extract values
654 | article_id = article_item.get("sys_id", "")
655 | title = article_item.get("short_description", "")
656 |
657 | # Extract nested values safely
658 | knowledge_base = ""
659 | if isinstance(article_item.get("kb_knowledge_base"), dict):
660 | knowledge_base = article_item["kb_knowledge_base"].get("display_value", "")
661 |
662 | category = ""
663 | if isinstance(article_item.get("kb_category"), dict):
664 | category = article_item["kb_category"].get("display_value", "")
665 |
666 | workflow_state = ""
667 | if isinstance(article_item.get("workflow_state"), dict):
668 | workflow_state = article_item["workflow_state"].get("display_value", "")
669 |
670 | created = article_item.get("sys_created_on", "")
671 | updated = article_item.get("sys_updated_on", "")
672 |
673 | articles.append({
674 | "id": article_id,
675 | "title": title,
676 | "knowledge_base": knowledge_base,
677 | "category": category,
678 | "workflow_state": workflow_state,
679 | "created": created,
680 | "updated": updated,
681 | })
682 | else:
683 | logger.warning("Result is not a list: %s", result)
684 |
685 | return {
686 | "success": True,
687 | "message": f"Found {len(articles)} articles",
688 | "articles": articles,
689 | "count": len(articles),
690 | "limit": params.limit,
691 | "offset": params.offset,
692 | }
693 |
694 | except requests.RequestException as e:
695 | logger.error(f"Failed to list articles: {e}")
696 | return {
697 | "success": False,
698 | "message": f"Failed to list articles: {str(e)}",
699 | "articles": [],
700 | "count": 0,
701 | "limit": params.limit,
702 | "offset": params.offset,
703 | }
704 |
705 |
706 | def get_article(
707 | config: ServerConfig,
708 | auth_manager: AuthManager,
709 | params: GetArticleParams,
710 | ) -> Dict[str, Any]:
711 | """
712 | Get a specific knowledge article by ID.
713 |
714 | Args:
715 | config: Server configuration.
716 | auth_manager: Authentication manager.
717 | params: Parameters for getting the article.
718 |
719 | Returns:
720 | Dictionary with article details.
721 | """
722 | api_url = f"{config.api_url}/table/kb_knowledge/{params.article_id}"
723 |
724 | # Build query parameters
725 | query_params = {
726 | "sysparm_display_value": "true",
727 | }
728 |
729 | # Make request
730 | try:
731 | response = requests.get(
732 | api_url,
733 | params=query_params,
734 | headers=auth_manager.get_headers(),
735 | timeout=config.timeout,
736 | )
737 | response.raise_for_status()
738 |
739 | # Get the JSON response
740 | json_response = response.json()
741 |
742 | # Safely extract the result
743 | if isinstance(json_response, dict) and "result" in json_response:
744 | result = json_response.get("result", {})
745 | else:
746 | logger.error("Unexpected response format: %s", json_response)
747 | return {
748 | "success": False,
749 | "message": "Unexpected response format",
750 | }
751 |
752 | if not result or not isinstance(result, dict):
753 | return {
754 | "success": False,
755 | "message": f"Article with ID {params.article_id} not found",
756 | }
757 |
758 | # Extract values safely
759 | article_id = result.get("sys_id", "")
760 | title = result.get("short_description", "")
761 | text = result.get("text", "")
762 |
763 | # Extract nested values safely
764 | knowledge_base = ""
765 | if isinstance(result.get("kb_knowledge_base"), dict):
766 | knowledge_base = result["kb_knowledge_base"].get("display_value", "")
767 |
768 | category = ""
769 | if isinstance(result.get("kb_category"), dict):
770 | category = result["kb_category"].get("display_value", "")
771 |
772 | workflow_state = ""
773 | if isinstance(result.get("workflow_state"), dict):
774 | workflow_state = result["workflow_state"].get("display_value", "")
775 |
776 | author = ""
777 | if isinstance(result.get("author"), dict):
778 | author = result["author"].get("display_value", "")
779 |
780 | keywords = result.get("keywords", "")
781 | article_type = result.get("article_type", "")
782 | views = result.get("view_count", "0")
783 | created = result.get("sys_created_on", "")
784 | updated = result.get("sys_updated_on", "")
785 |
786 | article = {
787 | "id": article_id,
788 | "title": title,
789 | "text": text,
790 | "knowledge_base": knowledge_base,
791 | "category": category,
792 | "workflow_state": workflow_state,
793 | "created": created,
794 | "updated": updated,
795 | "author": author,
796 | "keywords": keywords,
797 | "article_type": article_type,
798 | "views": views,
799 | }
800 |
801 | return {
802 | "success": True,
803 | "message": "Article retrieved successfully",
804 | "article": article,
805 | }
806 |
807 | except requests.RequestException as e:
808 | logger.error(f"Failed to get article: {e}")
809 | return {
810 | "success": False,
811 | "message": f"Failed to get article: {str(e)}",
812 | }
813 |
814 |
815 | def list_categories(
816 | config: ServerConfig,
817 | auth_manager: AuthManager,
818 | params: ListCategoriesParams,
819 | ) -> Dict[str, Any]:
820 | """
821 | List categories in a knowledge base.
822 |
823 | Args:
824 | config: Server configuration.
825 | auth_manager: Authentication manager.
826 | params: Parameters for listing categories.
827 |
828 | Returns:
829 | Dictionary with list of categories and metadata.
830 | """
831 | api_url = f"{config.api_url}/table/kb_category"
832 |
833 | # Build query parameters
834 | query_params = {
835 | "sysparm_limit": params.limit,
836 | "sysparm_offset": params.offset,
837 | "sysparm_display_value": "all",
838 | }
839 |
840 | # Build query string
841 | query_parts = []
842 | if params.knowledge_base:
843 | # Try different query format to ensure we match by sys_id value
844 | query_parts.append(f"kb_knowledge_base.sys_id={params.knowledge_base}")
845 | if params.parent_category:
846 | query_parts.append(f"parent.sys_id={params.parent_category}")
847 | if params.active is not None:
848 | query_parts.append(f"active={str(params.active).lower()}")
849 | if params.query:
850 | query_parts.append(f"labelLIKE{params.query}^ORdescriptionLIKE{params.query}")
851 |
852 | if query_parts:
853 | query_string = "^".join(query_parts)
854 | logger.debug(f"Constructed query string: {query_string}")
855 | query_params["sysparm_query"] = query_string
856 |
857 | # Log the query parameters for debugging
858 | logger.debug(f"Listing categories with query params: {query_params}")
859 |
860 | # Make request
861 | try:
862 | response = requests.get(
863 | api_url,
864 | params=query_params,
865 | headers=auth_manager.get_headers(),
866 | timeout=config.timeout,
867 | )
868 | response.raise_for_status()
869 |
870 | # Get the JSON response
871 | json_response = response.json()
872 |
873 | # Safely extract the result
874 | if isinstance(json_response, dict) and "result" in json_response:
875 | result = json_response.get("result", [])
876 | else:
877 | logger.error("Unexpected response format: %s", json_response)
878 | return {
879 | "success": False,
880 | "message": "Unexpected response format",
881 | "categories": [],
882 | "count": 0,
883 | "limit": params.limit,
884 | "offset": params.offset,
885 | }
886 |
887 | # Transform the results
888 | categories = []
889 |
890 | # Handle either string or list
891 | if isinstance(result, list):
892 | for category_item in result:
893 | if not isinstance(category_item, dict):
894 | logger.warning("Skipping non-dictionary category item: %s", category_item)
895 | continue
896 |
897 | # Safely extract values
898 | category_id = category_item.get("sys_id", "")
899 | title = category_item.get("label", "")
900 | description = category_item.get("description", "")
901 |
902 | # Extract knowledge base - handle both dictionary and string cases
903 | knowledge_base = ""
904 | kb_field = category_item.get("kb_knowledge_base")
905 | if isinstance(kb_field, dict):
906 | knowledge_base = kb_field.get("display_value", "")
907 | elif isinstance(kb_field, str):
908 | knowledge_base = kb_field
909 | # Also check if kb_knowledge_base is missing but there's a separate value field
910 | elif "kb_knowledge_base_value" in category_item:
911 | knowledge_base = category_item.get("kb_knowledge_base_value", "")
912 | elif "kb_knowledge_base.display_value" in category_item:
913 | knowledge_base = category_item.get("kb_knowledge_base.display_value", "")
914 |
915 | # Extract parent category - handle both dictionary and string cases
916 | parent = ""
917 | parent_field = category_item.get("parent")
918 | if isinstance(parent_field, dict):
919 | parent = parent_field.get("display_value", "")
920 | elif isinstance(parent_field, str):
921 | parent = parent_field
922 | # Also check alternative field names
923 | elif "parent_value" in category_item:
924 | parent = category_item.get("parent_value", "")
925 | elif "parent.display_value" in category_item:
926 | parent = category_item.get("parent.display_value", "")
927 |
928 | # Convert active to boolean - handle string or boolean types
929 | active_field = category_item.get("active")
930 | if isinstance(active_field, str):
931 | active = active_field.lower() == "true"
932 | elif isinstance(active_field, bool):
933 | active = active_field
934 | else:
935 | active = False
936 |
937 | created = category_item.get("sys_created_on", "")
938 | updated = category_item.get("sys_updated_on", "")
939 |
940 | categories.append({
941 | "id": category_id,
942 | "title": title,
943 | "description": description,
944 | "knowledge_base": knowledge_base,
945 | "parent_category": parent,
946 | "active": active,
947 | "created": created,
948 | "updated": updated,
949 | })
950 |
951 | # Log for debugging purposes
952 | logger.debug(f"Processed category: {title}, KB: {knowledge_base}, Parent: {parent}")
953 | else:
954 | logger.warning("Result is not a list: %s", result)
955 |
956 | return {
957 | "success": True,
958 | "message": f"Found {len(categories)} categories",
959 | "categories": categories,
960 | "count": len(categories),
961 | "limit": params.limit,
962 | "offset": params.offset,
963 | }
964 |
965 | except requests.RequestException as e:
966 | logger.error(f"Failed to list categories: {e}")
967 | return {
968 | "success": False,
969 | "message": f"Failed to list categories: {str(e)}",
970 | "categories": [],
971 | "count": 0,
972 | "limit": params.limit,
973 | "offset": params.offset,
974 | }
```