"""
ServiceNow MCP Server
This module provides the main implementation of the ServiceNow MCP server.
"""
import argparse
import os
from typing import Any, Dict, Union
import uvicorn
from dotenv import load_dotenv
from mcp.server import Server
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.routing import Mount, Route
from starlette.responses import JSONResponse
from servicenow_mcp.server import ServiceNowMCP
from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig
class AuthorizationMiddleware(BaseHTTPMiddleware):
def __init__(self, app, api_key):
super().__init__(app)
self.api_key = api_key
async def dispatch(self, request, call_next):
# Check for API key in Authorization header (Bearer token)
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return JSONResponse(
status_code=401,
content={"error": "Missing or invalid Authorization header"}
)
provided_token = auth_header.split(" ", 1)[1]
if provided_token != self.api_key:
return JSONResponse(
status_code=401,
content={"error": "Invalid API token"}
)
return await call_next(request)
def create_starlette_app(mcp_server: Server, *, debug: bool = False, middleware: Any | None = None) -> Starlette:
"""Create a Starlette application that can serve the provided mcp server with SSE."""
sse = SseServerTransport("/messages/")
async def handle_sse(request: Request) -> None:
async with sse.connect_sse(
request.scope,
request.receive,
request._send, # noqa: SLF001
) as (read_stream, write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options(),
)
if middleware is None:
middleware = []
return Starlette(
debug=debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
middleware=middleware,
)
class UpdatedServiceNowMCP(ServiceNowMCP):
"""
ServiceNow MCP Server implementation.
This class provides a Model Context Protocol (MCP) server for ServiceNow,
allowing LLMs to interact with ServiceNow data and functionality.
"""
def __init__(self, config: Union[Dict, ServerConfig]):
"""
Initialize the ServiceNow MCP server.
Args:
config: Server configuration, either as a dictionary or ServerConfig object.
"""
super().__init__(config)
# self.mcp_server = FastMCP("ServiceNow", port=8080, host="localhost")
def start(self, host: str = "0.0.0.0", port: int = 8080, api_key: str | None = None):
"""
Start the MCP server with SSE transport using Starlette and Uvicorn.
Args:
host: Host address to bind to
port: Port to listen on
"""
# Create Starlette app with SSE transport
middleware = [Middleware(AuthorizationMiddleware, api_key=api_key)]
starlette_app = create_starlette_app(self.mcp_server._mcp_server, debug=True, middleware=middleware)
# Run using uvicorn
uvicorn.run(starlette_app, host=host, port=port)
def create_servicenow_mcp(instance_url: str, username: str, password: str):
"""
Create a ServiceNow MCP server with minimal configuration.
This is a simplified factory function that creates a pre-configured
ServiceNow MCP server with basic authentication.
Args:
instance_url: ServiceNow instance URL
username: ServiceNow username
password: ServiceNow password
Returns:
A configured ServiceNowMCP instance ready to use
Example:
```python
from servicenow_mcp.server import create_servicenow_mcp
# Create an MCP server for ServiceNow
mcp = create_servicenow_mcp(
instance_url="https://instance.service-now.com",
username="admin",
password="password"
)
# Start the server
mcp.start()
```
"""
# Create basic auth config
auth_config = AuthConfig(
type=AuthType.BASIC, basic=BasicAuthConfig(username=username, password=password)
)
# Create server config
config = ServerConfig(instance_url=instance_url, auth=auth_config)
# Create and return server
return UpdatedServiceNowMCP(config)
def main():
load_dotenv(override=True)
# Parse command line arguments
parser = argparse.ArgumentParser(description="Run ServiceNow MCP SSE-based server")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8080, help="Port to listen on")
args = parser.parse_args()
server = create_servicenow_mcp(
instance_url=os.getenv("SERVICENOW_INSTANCE_URL"),
username=os.getenv("SERVICENOW_USERNAME"),
password=os.getenv("SERVICENOW_PASSWORD"),
)
server.start(host=args.host, port=args.port, api_key=os.getenv("SERVER_API_KEY"))
if __name__ == "__main__":
main()
