Python Examples

import requests
from datetime import datetime, timedelta

API_URL = "https://api.chatelier.net"

def chatelier_task(skill_id: str, input_data: dict) -> dict:
    """Execute a Chatelier API task."""
    response = requests.post(
        f"{API_URL}/a2a/tasks",
        json={"skillId": skill_id, "input": input_data}
    )
    response.raise_for_status()
    return response.json()["task"]["output"]


def search_hotels(location: str, checkin: str, checkout: str, guests: int = 2):
    """Search for available hotels."""
    result = chatelier_task("search_hotels", {
        "location": location,
        "checkin": checkin,
        "checkout": checkout,
        "guests": guests
    })
    return result["hotels"]


def check_availability(hotel_id: str, checkin: str, checkout: str, guests: int = 2):
    """Get available rooms for a specific hotel."""
    result = chatelier_task("check_availability", {
        "hotel_id": hotel_id,
        "checkin": checkin,
        "checkout": checkout,
        "guests": guests
    })
    return result["rooms"]


def create_booking(
    hotel_id: str,
    room_type_id: str,
    checkin: str,
    checkout: str,
    guest_name: str,
    guest_email: str,
    agent_wallet: str = None
):
    """Create a booking reservation."""
    input_data = {
        "hotel_id": hotel_id,
        "room_type_id": room_type_id,
        "checkin": checkin,
        "checkout": checkout,
        "guest": {
            "name": guest_name,
            "email": guest_email
        }
    }
    
    # Include agent referral for commission
    if agent_wallet:
        input_data["agent_referral"] = {
            "agent_id": "my-python-agent",
            "agent_name": "My Travel Bot",
            "wallet_address": agent_wallet
        }
    
    return chatelier_task("create_booking", input_data)


def confirm_payment(booking_id: str):
    """Check if payment has been received."""
    return chatelier_task("confirm_payment", {"booking_id": booking_id})


# ============================================
# Example Usage
# ============================================

if __name__ == "__main__":
    # Set dates (example: 2 weeks from now, 2 nights)
    checkin = (datetime.now() + timedelta(days=14)).strftime("%Y-%m-%d")
    checkout = (datetime.now() + timedelta(days=16)).strftime("%Y-%m-%d")
    
    print(f"🔍 Searching hotels in Sri Lanka for {checkin} to {checkout}...")
    
    # 1. Search hotels
    hotels = search_hotels("Sri Lanka", checkin, checkout, guests=2)
    print(f"   Found {len(hotels)} hotels")
    
    for hotel in hotels:
        print(f"   - {hotel['name']} (\${hotel['price']['from']}/night)")
    
    # 2. Check availability for first hotel
    hotel = hotels[0]
    print(f"\n📋 Checking availability at {hotel['name']}...")
    
    rooms = check_availability(hotel["slug"], checkin, checkout)
    print(f"   Found {len(rooms)} available room types")
    
    for room in rooms:
        print(f"   - {room['name']}: \${room['price']['total']} total")
    
    # 3. Create booking
    room = rooms[0]
    print(f"\n📝 Creating booking for {room['name']}...")
    
    booking = create_booking(
        hotel_id=hotel["slug"],
        room_type_id=room["room_type_id"],
        checkin=checkin,
        checkout=checkout,
        guest_name="John Doe",
        guest_email="john@example.com",
        agent_wallet="0x1234567890abcdef1234567890abcdef12345678"  # Your wallet
    )
    
    print(f"   ✅ Booking created: {booking['confirmation_number']}")
    print(f"   💰 Amount: {booking['total_amount']} {booking['currency']}")
    print(f"   📬 Pay to: {booking['payment']['receiver']}")
    print(f"   🔗 Payment page: {booking['payment_url']}")
    
    # 4. After guest pays, confirm payment
    # (In production, you'd poll this or use webhooks)
    print(f"\n⏳ Waiting for payment...")
    print(f"   (Send {booking['total_amount']} USDC to {booking['payment']['receiver']})")
    
    # Uncomment to check payment status:
    # result = confirm_payment(booking["booking_id"])
    # if result["confirmed"]:
    #     print(f"✅ Payment confirmed! TX: {result['payment']['tx_hash']}")

import httpx
import asyncio

API_URL = "https://api.chatelier.net"

async def chatelier_task(skill_id: str, input_data: dict) -> dict:
    """Execute a Chatelier API task asynchronously."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{API_URL}/a2a/tasks",
            json={"skillId": skill_id, "input": input_data},
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()["task"]["output"]


async def search_and_book():
    """Search hotels and create a booking."""
    # Search
    result = await chatelier_task("search_hotels", {
        "location": "Sri Lanka",
        "checkin": "2025-01-15",
        "checkout": "2025-01-17"
    })
    hotels = result["hotels"]
    print(f"Found {len(hotels)} hotels")
    
    # Check availability
    rooms = await chatelier_task("check_availability", {
        "hotel_id": hotels[0]["slug"],
        "checkin": "2025-01-15",
        "checkout": "2025-01-17"
    })
    print(f"Found {len(rooms['rooms'])} rooms")
    
    # Create booking
    booking = await chatelier_task("create_booking", {
        "hotel_id": hotels[0]["slug"],
        "room_type_id": rooms["rooms"][0]["room_type_id"],
        "checkin": "2025-01-15",
        "checkout": "2025-01-17",
        "guest": {"name": "Jane Doe", "email": "jane@example.com"},
        "agent_referral": {
            "agent_id": "async-agent",
            "wallet_address": "0x..."
        }
    })
    
    print(f"Booking created: {booking['confirmation_number']}")
    return booking


if __name__ == "__main__":
    asyncio.run(search_and_book())

import requests
from typing import Optional

class ChatelierError(Exception):
    """Base exception for Chatelier API errors."""
    def __init__(self, code: str, message: str, suggestion: str = None):
        self.code = code
        self.message = message
        self.suggestion = suggestion
        super().__init__(f"{code}: {message}")


class ChatelierClient:
    def __init__(self, base_url: str = "https://api.chatelier.net"):
        self.base_url = base_url
        self.session = requests.Session()
    
    def task(self, skill_id: str, input_data: dict) -> dict:
        try:
            response = self.session.post(
                f"{self.base_url}/a2a/tasks",
                json={"skillId": skill_id, "input": input_data},
                timeout=30
            )
            
            # Check for HTTP errors
            if response.status_code >= 400:
                error_data = response.json().get("error", {})
                raise ChatelierError(
                    code=error_data.get("code", "UNKNOWN"),
                    message=error_data.get("message", response.text),
                    suggestion=error_data.get("suggestion")
                )
            
            result = response.json()
            task = result.get("task", {})
            
            # Check for task-level errors
            if task.get("status") == "failed":
                raise ChatelierError(
                    code="TASK_FAILED",
                    message=task.get("error", "Task failed")
                )
            
            return task.get("output", {})
            
        except requests.exceptions.Timeout:
            raise ChatelierError("TIMEOUT", "Request timed out")
        except requests.exceptions.ConnectionError:
            raise ChatelierError("CONNECTION_ERROR", "Could not connect to API")


# Usage
client = ChatelierClient()

try:
    hotels = client.task("search_hotels", {
        "location": "Sri Lanka",
        "checkin": "2025-01-15",
        "checkout": "2025-01-17"
    })
    print(f"Found {len(hotels['hotels'])} hotels")
    
except ChatelierError as e:
    print(f"API Error: {e.code} - {e.message}")
    if e.suggestion:
        print(f"Suggestion: {e.suggestion}")

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests

class HotelSearchInput(BaseModel):
    location: str = Field(description="City or country to search")
    checkin: str = Field(description="Check-in date (YYYY-MM-DD)")
    checkout: str = Field(description="Check-out date (YYYY-MM-DD)")
    guests: int = Field(default=2, description="Number of guests")


class ChatelierSearchTool(BaseTool):
    name: str = "chatelier_search"
    description: str = """Search for available hotels. Use this when a user wants to 
    find hotels in a specific location. Returns hotel names, prices, and availability."""
    args_schema: type[BaseModel] = HotelSearchInput
    
    def _run(self, location: str, checkin: str, checkout: str, guests: int = 2) -> str:
        response = requests.post(
            "https://api.chatelier.net/a2a/tasks",
            json={
                "skillId": "search_hotels",
                "input": {
                    "location": location,
                    "checkin": checkin,
                    "checkout": checkout,
                    "guests": guests
                }
            }
        )
        hotels = response.json()["task"]["output"]["hotels"]
        
        # Format for LLM consumption
        result = f"Found {len(hotels)} hotels in {location}:\n"
        for h in hotels:
            result += f"- {h['name']}: from ${h['price']['from']}/night\n"
        return result


# Usage with LangChain agent
# tool = ChatelierSearchTool()
# agent = create_react_agent(llm, [tool], prompt)