"""ServiceNow integration service."""
from typing import Any, Dict, List, Optional
import httpx
from mcp.server import FastMCP
from ..base_service import BaseService
from .settings import ServiceNowSettings
class ServiceNowMCP(BaseService):
"""ServiceNow integration service."""
def __init__(self):
"""Initialize ServiceNow service."""
super().__init__()
self.settings = ServiceNowSettings()
self.client: Optional[httpx.AsyncClient] = None
async def register(self, mcp_server: FastMCP) -> None:
"""Register ServiceNow service with MCP server."""
self.log_registration("ServiceNow")
@mcp_server.tool()
async def get_incidents(limit: int = 10) -> List[Dict[str, Any]]:
"""Get ServiceNow incidents."""
return await self._get_incidents(limit)
@mcp_server.tool()
async def create_incident(
short_description: str,
description: str,
priority: int = 3,
category: str = "inquiry"
) -> Dict[str, Any]:
"""Create a new ServiceNow incident."""
return await self._create_incident(
short_description, description, priority, category
)
@mcp_server.tool()
async def get_incident(incident_id: str) -> Dict[str, Any]:
"""Get a specific ServiceNow incident by ID."""
return await self._get_incident(incident_id)
@mcp_server.tool()
async def update_incident(
incident_id: str,
**kwargs: Any
) -> Dict[str, Any]:
"""Update a ServiceNow incident."""
return await self._update_incident(incident_id, **kwargs)
@mcp_server.tool()
async def get_change_requests(limit: int = 10) -> List[Dict[str, Any]]:
"""Get ServiceNow change requests."""
return await self._get_change_requests(limit)
self.log_registration_complete("ServiceNow")
async def _get_client(self) -> httpx.AsyncClient:
"""Get or create HTTP client."""
if not self.client:
if not all([self.settings.INSTANCE_URL, self.settings.USERNAME, self.settings.PASSWORD]):
raise ValueError("ServiceNow credentials not configured")
self.client = httpx.AsyncClient(
base_url=f"{self.settings.INSTANCE_URL}/api/now/{self.settings.API_VERSION}",
auth=(self.settings.USERNAME, self.settings.PASSWORD),
headers={"Content-Type": "application/json", "Accept": "application/json"},
timeout=self.settings.TIMEOUT
)
return self.client
async def _get_incidents(self, limit: int) -> List[Dict[str, Any]]:
"""Get ServiceNow incidents."""
try:
client = await self._get_client()
response = await client.get(
"/table/incident",
params={"sysparm_limit": limit, "sysparm_display_value": "true"}
)
response.raise_for_status()
data = response.json()
return data.get("result", [])
except Exception as e:
self.logger.error(f"Error fetching incidents: {e}")
return []
async def _create_incident(
self,
short_description: str,
description: str,
priority: int,
category: str
) -> Dict[str, Any]:
"""Create a new ServiceNow incident."""
try:
client = await self._get_client()
data = {
"short_description": short_description,
"description": description,
"priority": priority,
"category": category,
}
response = await client.post("/table/incident", json=data)
response.raise_for_status()
return response.json().get("result", {})
except Exception as e:
self.logger.error(f"Error creating incident: {e}")
return {"error": str(e)}
async def _get_incident(self, incident_id: str) -> Dict[str, Any]:
"""Get a specific ServiceNow incident."""
try:
client = await self._get_client()
response = await client.get(
f"/table/incident/{incident_id}",
params={"sysparm_display_value": "true"}
)
response.raise_for_status()
return response.json().get("result", {})
except Exception as e:
self.logger.error(f"Error fetching incident {incident_id}: {e}")
return {"error": str(e)}
async def _update_incident(self, incident_id: str, **kwargs: Any) -> Dict[str, Any]:
"""Update a ServiceNow incident."""
try:
client = await self._get_client()
response = await client.put(f"/table/incident/{incident_id}", json=kwargs)
response.raise_for_status()
return response.json().get("result", {})
except Exception as e:
self.logger.error(f"Error updating incident {incident_id}: {e}")
return {"error": str(e)}
async def _get_change_requests(self, limit: int) -> List[Dict[str, Any]]:
"""Get ServiceNow change requests."""
try:
client = await self._get_client()
response = await client.get(
"/table/change_request",
params={"sysparm_limit": limit, "sysparm_display_value": "true"}
)
response.raise_for_status()
data = response.json()
return data.get("result", [])
except Exception as e:
self.logger.error(f"Error fetching change requests: {e}")
return []
async def __aenter__(self):
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit."""
if self.client:
await self.client.aclose()
