import os
import json
from starlette.applications import Starlette
from starlette.responses import JSONResponse, StreamingResponse
from starlette.routing import Route
from starlette.requests import Request
from mcp.server import Server
from mcp.types import Tool, TextContent
from .zendesk_client import ZendeskClient
API_KEY = os.environ.get("API_KEY")
def check_auth(request: Request) -> bool:
if not API_KEY:
return True
auth = request.headers.get("Authorization", "")
return auth == f"Bearer {API_KEY}"
async def health(request: Request):
return JSONResponse({"status": "ok", "service": "zendesk-mcp-server"})
async def sse_endpoint(request: Request):
if not check_auth(request):
return JSONResponse({"error": "Unauthorized"}, status_code=401)
async def event_stream():
# Send initial connection message
yield f"data: {json.dumps({'type': 'connected'})}\n\n"
# Keep connection alive
while True:
yield f": keepalive\n\n"
await asyncio.sleep(30)
return StreamingResponse(event_stream(), media_type="text/event-stream")
async def tools_endpoint(request: Request):
if not check_auth(request):
return JSONResponse({"error": "Unauthorized"}, status_code=401)
tools = [
{"name": "get_tickets", "description": "Get Zendesk tickets"},
{"name": "get_ticket", "description": "Get a specific ticket"},
{"name": "get_ticket_comments", "description": "Get ticket comments"},
{"name": "create_ticket", "description": "Create a new ticket"},
{"name": "create_ticket_comment", "description": "Add comment to ticket"},
{"name": "update_ticket", "description": "Update ticket properties"},
]
return JSONResponse({"tools": tools})
async def call_tool_endpoint(request: Request):
if not check_auth(request):
return JSONResponse({"error": "Unauthorized"}, status_code=401)
try:
body = await request.json()
name = body.get("name")
arguments = body.get("arguments", {})
client = ZendeskClient()
if name == "get_tickets":
result = await client.get_tickets(**arguments)
elif name == "get_ticket":
result = await client.get_ticket(arguments["ticket_id"])
elif name == "get_ticket_comments":
result = await client.get_ticket_comments(arguments["ticket_id"])
elif name == "create_ticket":
result = await client.create_ticket(**arguments)
elif name == "create_ticket_comment":
result = await client.create_ticket_comment(**arguments)
elif name == "update_ticket":
result = await client.update_ticket(**arguments)
else:
return JSONResponse({"error": f"Unknown tool: {name}"}, status_code=400)
return JSONResponse({"result": result})
except Exception as e:
return JSONResponse({"error": str(e)}, status_code=500)
import asyncio
routes = [
Route("/health", health),
Route("/sse", sse_endpoint),
Route("/tools", tools_endpoint),
Route("/call", call_tool_endpoint, methods=["POST"]),
]
app = Starlette(routes=routes)
