Catering Discovery Agent

pip install requests python-dotenv

import json
import os
from typing import Any, Dict, Optional

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
}

SYSTEM_PROMPT = """
You are Caterly — a professional event catering discovery assistant. Your task is to find, evaluate, and recommend caterers that best match a user's event requirements. Always prioritize accuracy, transparency, and user-safety.

Behavior rules:
1. Ask only necessary clarifying questions (if absolutely needed) — otherwise assume missing common defaults (guest_count=50, budget_per_person=$30, date=flexible).
2. When search_enabled is true, perform web lookups and verify vendor contact and availability information; include citation for each vendor where possible.
3. For each recommended caterer, provide: name, short description, service types, price per person or range, contact info, menu highlights, dietary accommodations, travel/fee considerations, and a confidence rating with explanation.
4. Quote prices clearly and compute total estimated event cost = guest_count * price_per_person + service/travel fees.
5. Provide a "next steps" checklist and a ready-to-send caterer outreach message.
6. If asked to make bookings, provide only a booking proposal — never execute payments or contact vendors unless a tool is explicitly connected.
7. Never invent contact data — mark unverifiable items explicitly as "unverified".
8. If a vendor is unavailable, offer alternatives or timeline adjustments.
9. Output structured JSON followed by a short human summary.
"""

def create_agent_payload(task: str, search_enabled: bool = True) -> Dict[str, Any]:
    return {
        "agent_config": {
            "agent_name": "Caterly - Event Catering Discovery Assistant",
            "description": (
                "Professional event catering discovery assistant that finds, "
                "evaluates, and recommends caterers matching your event "
                "requirements. Provides detailed vendor information including "
                "pricing, dietary accommodations, contact details, menu "
                "highlights, and ready-to-send outreach messages. Performs web "
                "searches to verify vendor availability and contact information "
                "when enabled."
            ),
            "system_prompt": SYSTEM_PROMPT,
            "model_name": "gpt-4.1",
            "max_tokens": 3000,
            "temperature": 0.5,
            "role": "worker",
            "max_loops": 1,
            "search_enabled": search_enabled,
            "tool_call_summary": True,
            "dynamic_temperature_enabled": True,
        },
        "task": task,
        "search_enabled": search_enabled,
    }

def run_caterly(task: str, search_enabled: bool = True) -> Dict[str, Any]:
    payload = create_agent_payload(task, search_enabled=search_enabled)
    response = requests.post(
        f"{BASE_URL}/v1/agent/completions",
        headers=headers,
        json=payload,
        timeout=300,
    )
    response.raise_for_status()
    return response.json()

def extract_content_from_outputs(outputs: Any) -> str:
    """Extract text content from the various output shapes the API returns."""
    if outputs is None:
        return ""
    if isinstance(outputs, str):
        return outputs
    if isinstance(outputs, list):
        parts = []
        for item in outputs:
            if isinstance(item, dict):
                content = item.get("content") or item.get("text") or item.get("message")
                if content:
                    parts.append(str(content))
            elif isinstance(item, str):
                parts.append(item)
        return "\n\n".join(parts) if parts else str(outputs)
    if isinstance(outputs, dict):
        return str(
            outputs.get("content")
            or outputs.get("text")
            or outputs.get("message")
            or json.dumps(outputs, indent=2)
        )
    return str(outputs)


def split_json_and_summary(content: str) -> Dict[str, Optional[str]]:
    """Caterly emits JSON first, then a human summary. Split them."""
    import re

    match = re.search(r"\{[\s\S]*\}", content)
    raw_json = match.group(0) if match else None
    parsed: Optional[Any] = None
    summary = content
    if raw_json:
        try:
            parsed = json.loads(raw_json)
            summary = content.replace(raw_json, "").strip()
        except json.JSONDecodeError:
            parsed = None
    return {"json": parsed, "summary": summary}

if __name__ == "__main__":
    task = (
        "Find 5 caterers for a 70 person event in Mission, San Francisco. "
        "Prioritize a pizza caterer that can deliver to the event location. "
        "For each vendor, provide name, address, phone number, email, "
        "website, and a brief description of their services. Compute the "
        "total estimated event cost for each option."
    )

    result = run_caterly(task, search_enabled=True)

    content = extract_content_from_outputs(result.get("outputs", ""))
    parsed = split_json_and_summary(content)

    print("\n=== STRUCTURED VENDORS ===")
    if parsed["json"]:
        print(json.dumps(parsed["json"], indent=2))
    else:
        print("(no JSON block found — model drifted from the contract)")

    print("\n=== HUMAN SUMMARY ===")
    print(parsed["summary"])

    usage = result.get("usage", {})
    print(f"\nTokens: {usage.get('total_tokens', 0)}  "
          f"Cost: ${usage.get('total_cost', 0):.4f}")