Temporal

Event 1: WorkflowStarted
Event 2: ActivityScheduled  (scrape_url, url1)
Event 3: ActivityCompleted  (scrape_url, url1) → "scraped text"
Event 4: SignalReceived      (RECEIVE_EVENT)
Event 5: ActivityScheduled  (scrape_url, url2)
          ← worker crashes here

import os
from pathlib import Path
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent / ".env")

from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.types.fastacp import TemporalACPConfig

acp = FastACP.create(
    acp_type="async",
    config=TemporalACPConfig(
        type="temporal",
        temporal_address=os.getenv("TEMPORAL_ADDRESS", "localhost:7233"),
    ),
)

from typing import override
from temporalio import workflow
from agentex.lib.core.temporal.types.workflow import SignalName
from agentex.lib.core.temporal.workflows.workflow import BaseWorkflow
from agentex.lib.environment_variables import EnvironmentVariables
from agentex.lib.types.acp import CreateTaskParams, SendEventParams

environment_variables = EnvironmentVariables.refresh()

@workflow.defn(name=environment_variables.WORKFLOW_NAME)
class MyWorkflow(BaseWorkflow):

    def __init__(self):
        super().__init__(display_name="My Agent")
        self._done = False  # set True to exit; usually stays False (cancelled externally)

    @workflow.run
    @override
    async def on_task_create(self, params: CreateTaskParams) -> None:
        # Called ONCE when the task is created.
        # Initialize state, send opening message, then block.
        await adk.state.create(task_id=params.task.id, agent_id=params.agent.id, state=MyState.initial())
        await adk.messages.create(task_id=params.task.id, content=TextContent(author="agent", content="Ready!"))
        await workflow.wait_condition(lambda: self._done)  # keeps workflow alive

    @workflow.signal(name=SignalName.RECEIVE_EVENT)
    @override
    async def on_task_event_send(self, params: SendEventParams) -> None:
        # Called on EVERY user message. Runs as a Temporal signal handler.
        # All logic for responding to user input lives here.
        ...

import httpx
from pydantic import BaseModel
from temporalio import activity

SCRAPE_URL_ACTIVITY = "scrape_url"  # string name must match workflow.execute_activity() call

class ScrapeURLParams(BaseModel):
    url: str  # serialized to JSON by Temporal when dispatching to the worker

class ScraperActivities:
    @activity.defn(name=SCRAPE_URL_ACTIVITY)
    async def scrape_url(self, params: ScrapeURLParams) -> str:
        async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client:
            response = await client.get(params.url)
            response.raise_for_status()  # non-2xx → exception → Temporal retries
        return response.text[:8000]

from datetime import timedelta
from temporalio.common import RetryPolicy

result: str = await workflow.execute_activity(
    SCRAPE_URL_ACTIVITY,
    ScrapeURLParams(url=url),
    start_to_close_timeout=timedelta(minutes=2),  # must finish within this window
    retry_policy=RetryPolicy(maximum_attempts=2),  # 2 total attempts before raising
)

# models.py
from agentex.lib.utils.model_utils import BaseModel

class MyState(BaseModel):
    turn: int = 0
    pending_urls: list[str] = []

    @classmethod
    def initial(cls) -> "MyState":
        return cls()

# Inside on_task_create
await adk.state.create(task_id=task_id, agent_id=agent_id, state=MyState.initial())

# Inside on_task_event_send
task_state = await adk.state.get_by_task_and_agent(task_id=task_id, agent_id=agent_id)
state = MyState.model_validate(task_state.state)   # deserialize

state.turn += 1                                     # mutate in-memory

await adk.state.update(                             # persist
    state_id=task_state.id,
    task_id=task_id,
    agent_id=agent_id,
    state=state,
)

import asyncio
from dotenv import load_dotenv
from pathlib import Path
load_dotenv(Path(__file__).parent / ".env")

from agentex.lib.core.temporal.activities import get_all_activities
from agentex.lib.core.temporal.workers.worker import AgentexWorker
from agentex.lib.environment_variables import EnvironmentVariables

from project.activities import ScraperActivities
from project.workflow import MyWorkflow

env = EnvironmentVariables.refresh()

async def main():
    scraper = ScraperActivities()
    worker = AgentexWorker(task_queue=env.WORKFLOW_TASK_QUEUE, health_check_port=8084)
    await worker.run(
        activities=[*get_all_activities(), scraper.scrape_url],
        workflow=MyWorkflow,
    )

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

try:
    page_text = await workflow.execute_activity(
        SCRAPE_URL_ACTIVITY, ScrapeURLParams(url=u),
        start_to_close_timeout=timedelta(minutes=2),
        retry_policy=RetryPolicy(maximum_attempts=2),
    )
    scraped_pages.append((u, page_text))
except Exception as e:
    await adk.messages.create(task_id=task_id,
        content=TextContent(author="agent", content=f"Failed to scrape `{u}`: {e}"))
    # continue loop — one bad URL doesn't abort the batch

try:
    task_state = await adk.state.get_by_task_and_agent(task_id=task_id, agent_id=agent_id)
    state = MyState.model_validate(task_state.state)
except Exception as e:
    await adk.messages.create(task_id=task_id,
        content=TextContent(author="agent", content=f"Failed to load state: {e}. Try again."))
    return

try:
    chat_completion = await adk.providers.litellm.chat_completion(llm_config=..., trace_id=task_id)
except Exception as e:
    await adk.messages.create(task_id=task_id,
        content=TextContent(author="agent", content=f"Summarization failed: {e}. Please resend URLs."))
    await adk.state.update(state_id=task_state.id, task_id=task_id, agent_id=agent_id, state=state)
    return