server.py•3.6 kB
import os
import re
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from requests_oauthlib import OAuth1Session
NETSUITE_ACCOUNT_ID = os.getenv("NETSUITE_ACCOUNT_ID")
NETSUITE_CONSUMER_KEY = os.getenv("NETSUITE_CONSUMER_KEY")
NETSUITE_CONSUMER_SECRET = os.getenv("NETSUITE_CONSUMER_SECRET")
NETSUITE_TOKEN_ID = os.getenv("NETSUITE_TOKEN_ID")
NETSUITE_TOKEN_SECRET = os.getenv("NETSUITE_TOKEN_SECRET")
if not all([
NETSUITE_ACCOUNT_ID,
NETSUITE_CONSUMER_KEY,
NETSUITE_CONSUMER_SECRET,
NETSUITE_TOKEN_ID,
NETSUITE_TOKEN_SECRET
]):
raise RuntimeError("Missing NetSuite TBA environment variables")
BASE_URL = f"https://{NETSUITE_ACCOUNT_ID}.suitetalk.api.netsuite.com"
app = FastAPI()
@app.get("/health")
async def health():
return {"status": "ok"}
@app.get("/.well-known/mcp/manifest")
async def mcp_manifest():
return JSONResponse(content={
"name": "mcp-netsuite",
"version": "0.1.0",
"description": "MCP server for Oracle NetSuite",
"tools": [
{
"name": "suiteql_query",
"description": "Execute read-only SuiteQL queries",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
},
{
"name": "get_customer",
"description": "Get customer by internalId",
"inputSchema": {
"type": "object",
"properties": {
"internalId": {"type": "string"}
},
"required": ["internalId"]
}
}
]
})
@app.post("/tools/suiteql_query")
async def suiteql_query(request: Request):
body = await request.json()
query = body.get("query", "").strip()
if not query.upper().startswith("SELECT"):
raise HTTPException(status_code=400, detail="Only SELECT allowed")
if re.search(r"(INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|EXEC|CALL)", query, re.IGNORECASE):
raise HTTPException(status_code=400, detail="Forbidden SQL keywords")
oauth = OAuth1Session(
client_key=NETSUITE_CONSUMER_KEY,
client_secret=NETSUITE_CONSUMER_SECRET,
resource_owner_key=NETSUITE_TOKEN_ID,
resource_owner_secret=NETSUITE_TOKEN_SECRET,
signature_method="HMAC-SHA256"
)
response = oauth.post(
f"{BASE_URL}/services/rest/query/v1/suiteql",
json={"q": query},
headers={"Prefer": "transient"}
)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
@app.post("/tools/get_customer")
async def get_customer(request: Request):
body = await request.json()
internal_id = body.get("internalId")
if not internal_id:
raise HTTPException(status_code=400, detail="internalId required")
oauth = OAuth1Session(
client_key=NETSUITE_CONSUMER_KEY,
client_secret=NETSUITE_CONSUMER_SECRET,
resource_owner_key=NETSUITE_TOKEN_ID,
resource_owner_secret=NETSUITE_TOKEN_SECRET,
signature_method="HMAC-SHA256"
)
response = oauth.get(f"{BASE_URL}/services/rest/record/v1/customer/{internal_id}")
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()