Build a Crash-Proof AI Agent with Temporal and OpenAI

%pip install --quiet temporalio openai-agents httpx nest-asyncio pytz

import asyncio
from datetime import datetime, timedelta

import httpx
import nest_asyncio
import pytz
from agents import Agent, Runner
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.contrib import openai_agents
from temporalio.contrib.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin
from temporalio.worker import Worker

print("[OK] All imports successful")

async def make_api_call(state: str) -> dict:
    """Make API call to fetch weather alerts from National Weather Service."""
    headers = {"User-Agent": "Temporal-Agents-Workshop/1.0 (educational@example.com)"}
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(
            f"https://api.weather.gov/alerts/active/area/{state}",
            headers=headers
        )
        r.raise_for_status()
        data = r.json()
    return data

async def make_api_call_bug(state: str) -> dict:
    """Make API call with intentional bug - simulates a failing service."""
    raise Exception("Simulated API failure: Service temporarily unavailable")

@activity.defn(name="get_weather")
async def get_weather(state: str) -> dict:
    """Fetch active NWS alerts for a 2-letter US state code (e.g., 'CA')."""
    data = await make_api_call(state)

    # Swap to make_api_call_bug(state) to simulate a failure:
    # data = await make_api_call_bug(state)

    alerts = []
    for f in (data.get("features") or [])[:5]:
        p = f.get("properties", {})
        alerts.append({
            "event": p.get("event"),
            "headline": p.get("headline"),
            "severity": p.get("severity"),
            "area": p.get("areaDesc"),
        })

    return {"state": state.upper(), "count": len(alerts), "alerts": alerts}

print("[OK] Activity 'get_weather' defined")

TASK_QUEUE = "agents-sdk-queue"

@workflow.defn(sandboxed=False)  # Disable sandbox for Jupyter compatibility
class WeatherAgentWorkflow:
    @workflow.run
    async def run(self, user_query: str) -> str:
        # Create the Agent inside the workflow with tools and instructions
        agent = Agent(
            name="Weather Assistant",
            instructions=(
                "You are a helpful assistant that explains current weather alerts for U.S. states. "
            ),
            tools=[
                # Convert Temporal activity to Agent tool for durable execution
                openai_agents.workflow.activity_as_tool(
                    get_weather,
                    start_to_close_timeout=timedelta(seconds=10),
                )
            ],
        )

        # Run the agent — tool calls execute as Temporal activities
        result = await Runner().run(agent, user_query)

        return getattr(result, "final_output", str(result))

print(f"[OK] Workflow 'WeatherAgentWorkflow' defined (task queue: {TASK_QUEUE})")

async def run_worker():
    """Start a Temporal worker that listens for workflow and activity tasks."""
    client = await Client.connect(
        "localhost:7233",
        plugins=[
            OpenAIAgentsPlugin(
                model_params=ModelActivityParameters(
                    start_to_close_timeout=timedelta(seconds=30)
                )
            )
        ],
    )

    worker = Worker(
        client,
        task_queue=TASK_QUEUE,
        workflows=[WeatherAgentWorkflow],
        activities=[get_weather],
    )

    print(f"[OK] Worker started on task queue: {TASK_QUEUE}")
    print("   Listening for workflow and activity tasks...")
    await worker.run()

nest_asyncio.apply()
worker_task = asyncio.create_task(run_worker())
first_worker_task = worker_task
print(" Worker running in background")

async def run_solution():
    """Execute the weather agent workflow."""
    query = "What weather alerts are active in CA?"

    est = pytz.timezone('US/Eastern')
    now = datetime.now(est)
    workflow_id = f"weather-{now.strftime('%a-%b-%d-%I%M%S').lower()}est"

    client = await Client.connect(
        "localhost:7233",
        plugins=[OpenAIAgentsPlugin()]
    )

    print(f" Starting workflow: {workflow_id}")
    print(f" Query: {query}\n")

    handle = await client.start_workflow(
        WeatherAgentWorkflow.run,
        query,
        id=workflow_id,
        task_queue=TASK_QUEUE
    )

    print(f"[OK] Workflow started: {handle.id}")
    print(f"🔗 View in Temporal UI: http://localhost:8233/namespaces/default/workflows/{workflow_id}\n")
    print(" Waiting for agent response...\n")

    result = await handle.result()

    print("=" * 60)
    print(" Agent Response:")
    print("=" * 60)
    print(result)
    print("=" * 60)

try:
    loop = asyncio.get_running_loop()
    await run_solution()
except RuntimeError:
    asyncio.run(run_solution())

# Change this:
data = await make_api_call(state)

# To this:
data = await make_api_call_bug(state)

# Cancel the first worker and wait for it to stop
first_worker_task.cancel()
try:
    await first_worker_task
except asyncio.CancelledError:
    pass

# Now start the new worker with the fixed code
worker_task = asyncio.create_task(run_worker())
print("New worker started!")