"""
ShopSupport: single-step tool-call example
What Snowglobe sends:
CompletionRequest(
messages=[SnowglobeMessage(role="user", content="Where is order A1001?")]
)
What you must return:
CompletionFunctionOutputs(response="string to display")
Notes:
- One OpenAI call only. If a tool is requested, we execute it once
then render a final user-facing string ourselves.
"""
# 1) Imports and setup
from snowglobe.client import CompletionRequest, CompletionFunctionOutputs
from openai import OpenAI
import os, json, uuid
from typing import Dict, Any, Callable, Tuple
MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 2) Toy app data and business functions
ORDERS: Dict[str, Dict[str, Any]] = {
"A1001": {"status": "Shipped", "eta_days": 2, "carrier": "UPS"},
"A1002": {"status": "Processing", "eta_days": 5, "carrier": None},
}
TICKETS: Dict[str, Dict[str, Any]] = {}
def get_order_status(order_id: str) -> Dict[str, Any]:
"""Return order details or found=False."""
data = ORDERS.get(order_id.upper())
if not data:
return {"found": False, "order_id": order_id}
return {"found": True, "order_id": order_id.upper(), **data}
def create_ticket(email: str, subject: str, body: str) -> Dict[str, Any]:
"""Create a support ticket and return its id."""
ticket_id = f"T{uuid.uuid4().hex[:8].upper()}"
TICKETS[ticket_id] = {"email": email, "subject": subject, "body": body, "status": "open"}
return {"ticket_id": ticket_id, "status": "open"}
# 3) Renderers: turn tool JSON into a final user-facing string
def render_order_status(out: Dict[str, Any], _args: Dict[str, Any]) -> str:
if not out.get("found"):
return f"I could not find order {out.get('order_id')}."
eta = out["eta_days"]
carrier = out["carrier"] or "carrier pending"
return f"Order {out['order_id']}: {out['status']}. ETA about {eta} day(s) via {carrier}."
def render_ticket(out: Dict[str, Any], args: Dict[str, Any]) -> str:
return f"Ticket {out['ticket_id']} opened. We will email updates to {args.get('email')}."
# 4) Tool specs for the model and a small dispatch table
TOOL_SPECS = [
{
"type": "function",
"function": {
"name": "get_order_status",
"description": "Look up an order by id like A1234",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"],
},
},
},
{
"type": "function",
"function": {
"name": "create_ticket",
"description": "Open a support ticket",
"parameters": {
"type": "object",
"properties": {
"email": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"},
},
"required": ["email", "subject", "body"],
},
},
},
]
# name -> (callable, renderer)
TOOL_REGISTRY: Dict[str, Tuple[Callable[..., Dict[str, Any]], Callable[[Dict[str, Any], Dict[str, Any]], str]]] = {
"get_order_status": (get_order_status, render_order_status),
"create_ticket": (create_ticket, render_ticket),
}
# 5) System prompt for predictable routing
SYSTEM_PROMPT = (
"You are ShopSupport. "
"If the user asks about an order, call get_order_status. "
"If they want to open a support ticket, call create_ticket. "
"If no tool fits, answer directly and be concise."
)
# 6) The Snowglobe entry point
def process_scenario(request: CompletionRequest) -> CompletionFunctionOutputs:
"""
Entry point called by Snowglobe. Returns a plain string response.
"""
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages += request.to_openai_messages()
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOL_SPECS,
tool_choice="auto",
)
msg = resp.choices[0].message
tool_calls = getattr(msg, "tool_calls", None)
# No tool called: just return the model's text
if not tool_calls:
return CompletionFunctionOutputs(response=msg.content or "")
# Single step: handle the first tool call only
tc = tool_calls[0]
try:
args = json.loads(tc.function.arguments or "{}")
except Exception:
return CompletionFunctionOutputs(response="I could not parse the tool arguments.")
handler = TOOL_REGISTRY.get(tc.function.name)
if not handler:
return CompletionFunctionOutputs(response="I cannot run the requested tool.")
func, render = handler
try:
out = func(**args) # run the business function
text = render(out, args) # format a user-friendly message
return CompletionFunctionOutputs(response=text)
except TypeError:
return CompletionFunctionOutputs(response="The tool arguments were incomplete.")
except Exception:
return CompletionFunctionOutputs(response="The tool failed to run.")