server.py•1.84 kB
"""
ServiceNow CMDB MCP Server
"""
import os
import httpx
import asyncio
from typing import Optional, List, Dict, Any
from fastmcp import FastMCP
from dotenv import load_dotenv
mcp = FastMCP("servicenow_cmdb_mcp")
load_dotenv()
SN_URL = os.getenv("SERVICENOW_URL", "").rstrip("/")
SN_USER = os.getenv("SERVICENOW_USER", "")
SN_PASS = os.getenv("SERVICENOW_PASS", "")
@mcp.tool()
async def get_cmdb_table_data(
table_name: str,
query: Optional[str] = None,
fields: Optional[str] = None,
limit: int = 100,
offset: int = 0
) -> Dict[str, Any]:
"""Get data from a CMDB table with optional query filtering and field selection."""
if not SN_URL or not SN_USER or not SN_PASS:
raise RuntimeError("Missing ServiceNow credentials")
params = {
"sysparm_limit": str(limit),
"sysparm_offset": str(offset),
"sysparm_display_value": "true"
}
if query:
params["sysparm_query"] = query
if fields:
params["sysparm_fields"] = fields
headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
async with httpx.AsyncClient(auth=(SN_USER, SN_PASS)) as client:
response = await client.get(
f"{SN_URL}/api/now/table/{table_name}",
params=params,
headers=headers
)
try:
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
raise RuntimeError(f"ServiceNow HTTP error {response.status_code}: {response.text}") from e
except ValueError:
raise RuntimeError(f"Non-JSON response from ServiceNow: {response.text}")
if __name__ == "__main__":
asyncio.run(mcp.run(transport="http", host="127.0.0.1", port=9123, log_level="info"))