Spaces:
Sleeping
Sleeping
File size: 4,370 Bytes
35e37df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
from dotenv import load_dotenv
import requests
import time
import os
from typing import Optional, Dict, Any
load_dotenv()
# --- ServiceNow OAuth / instance config ---
INSTANCE_URL = os.getenv("SERVICENOW_INSTANCE_URL") # e.g., https://your-instance.service-now.com
CLIENT_ID = os.getenv("SERVICENOW_CLIENT_ID")
CLIENT_SECRET = os.getenv("SERVICENOW_CLIENT_SECRET")
# Optional but recommended: provide defaults for create payload (helps BR/ACL allow quick resolution)
DEFAULT_ASSIGNMENT_GROUP_SYSID = os.getenv("SERVICENOW_ASSIGNMENT_GROUP_SYSID") # e.g., Service Desk group sys_id
DEFAULT_CATEGORY = os.getenv("SERVICENOW_DEFAULT_CATEGORY", "inquiry")
DEFAULT_SUBCATEGORY = os.getenv("SERVICENOW_DEFAULT_SUBCATEGORY", "general")
# OAuth token storage (simple in-memory cache)
_token_info: Dict[str, Any] = {"access_token": None, "expires_at": 0}
# Read SSL verify flag from env (default True)
VERIFY_SSL = os.getenv("SERVICENOW_SSL_VERIFY", "true").lower() in ("1", "true", "yes")
def get_oauth_token() -> str:
"""
Fetch a new OAuth token using client credentials flow.
"""
if not INSTANCE_URL or not CLIENT_ID or not CLIENT_SECRET:
raise RuntimeError(
"ServiceNow OAuth env vars are missing (SERVICENOW_INSTANCE_URL / CLIENT_ID / CLIENT_SECRET)."
)
url = f"{INSTANCE_URL}/oauth_token.do"
data = {
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"scope": "global",
}
# NOTE: verify=False is OK for local dev; set verify=True in production
resp = requests.post(url, data=data, verify=VERIFY_SSL, timeout=25) # <-- use VERIFY_SSL
if resp.status_code == 200:
tok = resp.json()
_token_info["access_token"] = tok.get("access_token")
_token_info["expires_at"] = time.time() + tok.get("expires_in", 0)
return _token_info["access_token"]
raise Exception(f"Failed to get token: {resp.status_code} - {resp.text}")
def get_valid_token() -> str:
"""
Return a valid token, refreshing if expired.
"""
if _token_info["access_token"] and time.time() < _token_info["expires_at"]:
return _token_info["access_token"]
return get_oauth_token()
def create_incident(short_description: str, description: str) -> Dict[str, Any]:
"""
Create an incident in ServiceNow using OAuth token.
Returns:
- {"number": "<INC...>", "sys_id": "<sys_id>"} on success
- {"error": "<message>", "status_code": <int>} on failure
NOTE:
- Includes optional defaults (category/subcategory/assignment_group) to satisfy common BR/ACL requirements
so your subsequent PATCH to Resolved succeeds.
"""
token = get_valid_token()
url = f"{INSTANCE_URL}/api/now/table/incident"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
payload: Dict[str, Any] = {
"short_description": short_description,
"description": description,
"urgency": "3", # Medium
"priority": "3",
# Optional defaults—adjust for your instance if needed
"category": DEFAULT_CATEGORY,
"subcategory": DEFAULT_SUBCATEGORY,
"caller_id": os.getenv("SERVICENOW_CALLER_SYSID"),
}
# Only include assignment_group if the env var is set
if DEFAULT_ASSIGNMENT_GROUP_SYSID:
payload["assignment_group"] = DEFAULT_ASSIGNMENT_GROUP_SYSID
# NOTE: verify=False is OK for local dev; set verify=True in production
resp = requests.post(url, headers=headers, json=payload, verify=VERIFY_SSL, timeout=30) # <-- use VERIFY_SSL
if resp.status_code in (200, 201):
try:
result = resp.json().get("result", {})
except Exception:
# Some instances respond with bare object
result = resp.json()
number = result.get("number")
sys_id = result.get("sys_id")
if number and sys_id:
return {"number": number, "sys_id": sys_id}
# If SN responded but fields missing, surface an error
return {"error": f"Missing Incident fields in response: {resp.text}", "status_code": resp.status_code}
# Failure: return structured error for caller (main.py /incident & /chat will handle this gracefully)
|