संरचित आउटपुट

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": "Return 3 upcoming basketball events near San Francisco as JSON.",
        }],
        response_format=EventsResponse,
    )

    parsed: EventsResponse = completion.choices[0].message.parsed
    print(parsed)

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

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[
            {
                "role": "user",
                "content": (
                    "Return 3 upcoming basketball events near San Francisco as JSON. "
                    "Use ISO dates (YYYY-MM-DD)."
                ),
            }
        ],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],  # Ticketmaster के माध्यम से इवेंट खोजें
    )

    parsed = completion.choices[0].message.parsed
    print(parsed)

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

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    # स्ट्रीमिंग के लिए context manager का उपयोग करें
    async with client.chat.completions.stream(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "Return 3 upcoming basketball events near San Francisco as JSON. "
                "Use ISO dates (YYYY-MM-DD)."
            ),
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    ) as stream:
        # इवेंट्स को आते ही प्रोसेस करें
        async for event in stream:
            if event.type == "content.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "content.done":
                # content.done पर स्नैपशॉट उपलब्ध होता है (typed)
                print(f"\nParsed events: {len(event.parsed.events)}")

        # अंतिम पार्स किया हुआ परिणाम प्राप्त करें
        final = await stream.get_final_completion()
        parsed = final.choices[0].message.parsed
        print(f"\nFinal events: {len(parsed.events)}")

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

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str
    price_usd: int | None = None  # model unknown as null

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "Return 3 upcoming basketball events near San Francisco as JSON. "
                "Include price_usd if known; otherwise null. Use ISO dates (YYYY-MM-DD)."
            ),
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, e.price_usd)

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

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Venue(BaseModel):
    name: str
    address: str | None = None
    city: str

class Event(BaseModel):
    name: str
    date: str
    venue: Venue

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "सैन फ्रांसिस्को के पास होने वाले 3 आगामी बास्केटबॉल इवेंट्स को JSON के रूप में लौटाएँ। "
                "हर इवेंट में name, city और address (अज्ञात होने पर null) फ़ील्ड्स वाला एक nested venue object ज़रूर शामिल हो। "
                "तारीखों के लिए ISO फ़ॉर्मेट (YYYY-MM-DD) का उपयोग करें।"
            )
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, "→", e.venue.name)

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class SearchEventsArgs(BaseModel):
    city: str
    month: str
    max_results: int = 5

async def main():
    client = AsyncDedalus()

    tools = [
        {
            "type": "function",
            "function": {
                "name": "search_events",
                "description": "Search for events in a city during a month.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string"},
                        "month": {"type": "string", "description": "YYYY-MM"},
                        "max_results": {"type": "integer", "default": 5},
                    },
                    "required": ["city", "month"],
                    "additionalProperties": False,
                },
                "strict": True,
            }
        }
    ]

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": "Call search_events for San Francisco in 2026-01.",
        }],
        tools=tools,
        tool_choice={"type": "tool", "name": "search_events"},
    )

    message = completion.choices[0].message
    if message.tool_calls:
        tool_call = message.tool_calls[0]
        print(f"Tool called: {tool_call.function.name}")
        print(f"Parsed args: {tool_call.function.parsed_arguments}")

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

import asyncio
from typing import Literal
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str
    category: Literal["sports", "music", "theater", "other"]
    ticket_status: Literal["available", "sold_out", "unknown"]

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "San Francisco के पास होने वाले 3 आगामी इवेंट्स को JSON के रूप में लौटाएँ। "
                "प्रत्येक इवेंट में category (sports/music/theater/other) और ticket_status (available/sold_out/unknown) शामिल होना चाहिए। "
                "ISO तारीख़ फ़ॉर्मेट (YYYY-MM-DD) का उपयोग करें।"
            )
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, e.category, e.ticket_status)

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

import asyncio

from dedalus_labs import AsyncDedalus, DedalusRunner
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

def as_bullets(items: list[str]) -> str:
    """आइटमों को बुलेटेड सूची के रूप में फ़ॉर्मैट करें।"""
    return "\n".join(f"• {item}" for item in items)

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input=(
            "Ticketmaster का उपयोग करके जनवरी में San Francisco में मेरे लिए सबसे नज़दीकी बास्केटबॉल गेम ढूँढें। "
            "फिर as_bullets को आइटमों की एक सूची के साथ कॉल करें (प्रति इवेंट एक: नाम, शहर, दिनांक)।"
        ),
        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/ticketmaster-mcp"],  # Ticketmaster के माध्यम से इवेंट खोजें
        tools=[as_bullets],
        response_format=EventsResponse,
        max_steps=5,
    )

    print(result.final_output)

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

import asyncio
from typing import Any
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    try:
        completion = await client.chat.completions.parse(
            model="openai/gpt-5.2",
            messages=[{
                "role": "user",
                "content": (
                    "Return 3 upcoming basketball events near San Francisco as JSON. "
                    "Use ISO dates (YYYY-MM-DD)."
                ),
            }],
            response_format=EventsResponse,
        )
        parsed = completion.choices[0].message.parsed
        print(f"Parsed events: {len(parsed.events)}")
    except Exception as e:
        print("Parse failed:", e)

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