Client

WorkflowClient

Python client for interacting with a WorkflowServer.

Provides methods for listing workflows, running them synchronously or asynchronously, streaming events, and sending events for human-in-the-loop workflows.

Example:

from llama_agents.client import WorkflowClient
from workflows.events import StartEvent

client = WorkflowClient(base_url="http://localhost:8080")

# Run synchronously
result = await client.run_workflow("greet", start_event=StartEvent(name="Ada"))
print(result.result)

# Run async and stream events
handler = await client.run_workflow_nowait("greet")
stream = client.get_workflow_events(handler.handler_id)
async for event in stream:
    print(event.type, event.value)

Parameters:

Name	Type	Description	Default
base_url	str | None	Base URL of the workflow server (e.g. "http://localhost:8080").	None
httpx_client	AsyncClient | None	Pre-configured httpx.AsyncClient. Use this for custom auth headers, timeouts, or transport configuration.	None

Provide exactly one of base_url or httpx_client.

Source code in llama_agents/client/client.py

class WorkflowClient:
    """Python client for interacting with a ``WorkflowServer``.

    Provides methods for listing workflows, running them synchronously or
    asynchronously, streaming events, and sending events for
    human-in-the-loop workflows.

    Example:

        from llama_agents.client import WorkflowClient
        from workflows.events import StartEvent

        client = WorkflowClient(base_url="http://localhost:8080")

        # Run synchronously
        result = await client.run_workflow("greet", start_event=StartEvent(name="Ada"))
        print(result.result)

        # Run async and stream events
        handler = await client.run_workflow_nowait("greet")
        stream = client.get_workflow_events(handler.handler_id)
        async for event in stream:
            print(event.type, event.value)

    Args:
        base_url: Base URL of the workflow server (e.g. ``"http://localhost:8080"``).
        httpx_client: Pre-configured ``httpx.AsyncClient``. Use this for
            custom auth headers, timeouts, or transport configuration.

    Provide exactly one of ``base_url`` or ``httpx_client``.
    """

    @overload
    def __init__(self, *, httpx_client: httpx.AsyncClient): ...
    @overload
    def __init__(
        self,
        *,
        base_url: str,
    ): ...

    def __init__(
        self,
        *,
        httpx_client: httpx.AsyncClient | None = None,
        base_url: str | None = None,
    ):
        if httpx_client is None and base_url is None:
            raise ValueError("Either httpx_client or base_url must be provided")
        if httpx_client is not None and base_url is not None:
            raise ValueError("Only one of httpx_client or base_url must be provided")
        self.httpx_client = httpx_client
        self.base_url = base_url

    @asynccontextmanager
    async def _get_client(self) -> AsyncIterator[httpx.AsyncClient]:
        if self.httpx_client:
            yield self.httpx_client
        else:
            async with httpx.AsyncClient(base_url=self.base_url or "") as client:
                yield client

    async def is_healthy(self) -> HealthResponse:
        """Check whether the workflow server is healthy.

        Raises:
            httpx.HTTPStatusError: If the server returns an error status.
        """
        async with self._get_client() as client:
            response = await client.get("/health")
            _raise_for_status_with_body(response)
            return HealthResponse.model_validate(response.json())

    async def list_workflows(self) -> WorkflowsListResponse:
        """List the names of all workflows registered on the server."""
        async with self._get_client() as client:
            response = await client.get("/workflows")

            _raise_for_status_with_body(response)

            return WorkflowsListResponse.model_validate(response.json())

    async def get_workflow_schema(self, workflow_name: str) -> WorkflowSchemaResponse:
        """Get the start and stop event schemas for a workflow."""
        async with self._get_client() as client:
            response = await client.get(f"/workflows/{workflow_name}/schema")
            _raise_for_status_with_body(response)
            return WorkflowSchemaResponse.model_validate(response.json())

    async def get_workflow_events_schema(
        self, workflow_name: str
    ) -> WorkflowEventsListResponse:
        """Get the schemas of events registered for a workflow."""
        async with self._get_client() as client:
            response = await client.get(f"/workflows/{workflow_name}/events")
            _raise_for_status_with_body(response)
            return WorkflowEventsListResponse.model_validate(response.json())

    async def get_workflow_graph(self, workflow_name: str) -> WorkflowGraphResponse:
        """Get the directed graph representation of a workflow."""
        async with self._get_client() as client:
            response = await client.get(f"/workflows/{workflow_name}/representation")
            _raise_for_status_with_body(response)
            return WorkflowGraphResponse.model_validate(response.json())

    async def run_workflow(
        self,
        workflow_name: str,
        handler_id: str | None = None,
        start_event: StartEvent | dict[str, Any] | None = None,
        context: Context | dict[str, Any] | None = None,
    ) -> HandlerData:
        """Run the workflow and block until completion.

        Args:
            workflow_name: Name of the registered workflow to run.
            start_event: Input event for the workflow. Can be a ``StartEvent``
                instance or a plain dict.
            context: Workflow context to restore, for continuing a previous run.
            handler_id: Handler identifier to continue from a previous
                completed run.

        Returns:
            HandlerData: Handler metadata including the final result.
        """
        if start_event is not None:
            try:
                start_event = _serialize_event(start_event, bare=True)
            except Exception as e:
                raise ValueError(
                    f"Impossible to serialize the start event because of: {e}"
                )
        if isinstance(context, Context):
            try:
                context = context.to_dict()
            except Exception as e:
                raise ValueError(f"Impossible to serialize the context because of: {e}")
        request_body: dict[str, Any] = {
            "start_event": start_event
            if start_event is not None
            else _serialize_event(StartEvent(), bare=True),
            "context": context if context is not None else {},
        }
        if handler_id:
            request_body["handler_id"] = handler_id
        async with self._get_client() as client:
            response = await client.post(
                f"/workflows/{workflow_name}/run", json=request_body
            )

            _raise_for_status_with_body(response)

            return HandlerData.model_validate(response.json())

    async def run_workflow_nowait(
        self,
        workflow_name: str,
        handler_id: str | None = None,
        start_event: StartEvent | dict[str, Any] | None = None,
        context: Context | dict[str, Any] | None = None,
    ) -> HandlerData:
        """Start the workflow without waiting for completion.

        Use the returned ``handler_id`` to stream events, poll for results,
        or send events.

        Args:
            workflow_name: Name of the registered workflow to run.
            start_event: Input event for the workflow. Can be a ``StartEvent``
                instance or a plain dict.
            context: Workflow context to restore, for continuing a previous run.
            handler_id: Handler identifier to continue from a previous
                completed run.

        Returns:
            HandlerData: Handler metadata including the ``handler_id``.
        """
        if start_event is not None:
            try:
                start_event = _serialize_event(start_event)
            except Exception as e:
                raise ValueError(
                    f"Impossible to serialize the start event because of: {e}"
                )
        if isinstance(context, Context):
            try:
                context = context.to_dict()
            except Exception as e:
                raise ValueError(f"Impossible to serialize the context because of: {e}")
        request_body: dict[str, Any] = {
            "start_event": start_event
            if start_event is not None
            else _serialize_event(StartEvent()),
            "context": context if context is not None else {},
        }
        if handler_id:
            request_body["handler_id"] = handler_id
        async with self._get_client() as client:
            response = await client.post(
                f"/workflows/{workflow_name}/run-nowait", json=request_body
            )

            _raise_for_status_with_body(response)

            return HandlerData.model_validate(response.json())

    def get_workflow_events(
        self,
        handler_id: str,
        include_internal_events: bool = False,
        after_sequence: int | Literal["now"] = -1,
        max_reconnect_attempts: int = 3,
    ) -> EventStream:
        """Stream events as they are produced by the workflow.

        Returns an ``EventStream`` whose ``last_sequence`` property tracks
        the sequence number of the most recently yielded event. Uses SSE
        and automatically reconnects from the last received event on
        connection drops.

        Example:

            stream = client.get_workflow_events(handler_id)
            async for event in stream:
                print(event.type, stream.last_sequence)

        Args:
            handler_id: ID of the handler running the workflow.
            include_internal_events: Include internal dispatch events.
                Defaults to ``False``.
            after_sequence: Where to start streaming. ``-1`` (default) streams
                all events from the beginning. ``"now"`` skips existing events
                and only delivers new ones. An integer ``N`` streams events
                after sequence ``N``.
            max_reconnect_attempts: Maximum reconnect attempts on connection
                drop. Defaults to ``3``.
        """
        queue: asyncio.Queue[_QueueItem] = asyncio.Queue()
        stream = EventStream(queue, None, after_sequence)

        async def reader() -> None:
            incl_inter = "true" if include_internal_events else "false"
            url = f"/events/{handler_id}"
            last_sequence: int | Literal["now"] = after_sequence
            attempts = 0
            try:
                while True:
                    async with self._get_client() as client:
                        try:
                            async with client.stream(
                                "GET",
                                url,
                                params={
                                    "sse": "true",
                                    "include_internal": incl_inter,
                                    "after_sequence": str(last_sequence),
                                },
                                headers={"Connection": "keep-alive"},
                                timeout=None,
                            ) as response:
                                if response.status_code == 404:
                                    raise ValueError("Handler not found")
                                elif response.status_code == 204:
                                    await queue.put(_QueuedDone())
                                    return

                                _raise_for_status_with_body(response)

                                # Reset attempts on successful connection
                                attempts = 0

                                # Parse SSE stream: "id: N\ndata: {...}\n\n"
                                current_id: str | None = None
                                async for line in response.aiter_lines():
                                    stripped = line.strip()
                                    if not stripped:
                                        # Empty line = end of SSE event
                                        continue
                                    if stripped.startswith("id:"):
                                        current_id = stripped[3:].strip()
                                    elif stripped.startswith("data:"):
                                        data = stripped[5:].strip()
                                        event = EventEnvelopeWithMetadata.model_validate_json(
                                            data
                                        )
                                        if current_id is not None:
                                            try:
                                                last_sequence = int(current_id)
                                            except ValueError:
                                                pass
                                        await queue.put(
                                            _QueuedEvent(
                                                sequence=last_sequence,
                                                event=event,
                                            )
                                        )
                                        current_id = None

                            # Stream ended normally (server closed connection)
                            await queue.put(_QueuedDone())
                            return

                        except httpx.TimeoutException:
                            raise TimeoutError(
                                f"Timeout waiting for events from handler {handler_id}"
                            )
                        except (httpx.RequestError, ConnectionError):
                            attempts += 1
                            if attempts > max_reconnect_attempts:
                                raise ConnectionError(
                                    f"Failed to connect to event stream after {max_reconnect_attempts} attempts"
                                )
                            # Retry from last received sequence
            except asyncio.CancelledError:
                await queue.put(_QueuedDone())
            except BaseException as exc:
                await queue.put(_QueuedError(exc))

        stream._task = asyncio.create_task(reader())
        return stream

    async def send_event(
        self,
        handler_id: str,
        event: Event | dict[str, Any],
        step: str | None = None,
    ) -> SendEventResponse:
        """Send an event to a running workflow.

        Useful for human-in-the-loop workflows that wait for external input.

        Args:
            handler_id: ID of the handler running the workflow.
            event: Event to send, as an ``Event`` instance or a dict.
            step: Target a specific workflow step. When ``None``, the event
                is broadcast to all waiting steps.
        """
        try:
            serialized_event: dict[str, Any] = _serialize_event(event)
        except Exception as e:
            raise ValueError(f"Error while serializing the provided event: {e}")
        request_body: dict[str, Any] = {"event": serialized_event}
        if step:
            request_body["step"] = step
        async with self._get_client() as client:
            response = await client.post(f"/events/{handler_id}", json=request_body)
            _raise_for_status_with_body(response)

            return SendEventResponse.model_validate(response.json())

    async def get_result(self, handler_id: str) -> HandlerData:
        """
        Deprecated. Use get_handler instead.
        """
        return await self.get_handler(handler_id)

    async def get_handlers(
        self,
        status: list[Status] | None = None,
        workflow_name: list[str] | None = None,
    ) -> HandlersListResponse:
        """List all workflow handlers.

        Args:
            status: Filter by handler status (e.g. ``"running"``,
                ``"completed"``).
            workflow_name: Filter by workflow name.
        """
        async with self._get_client() as client:
            response = await client.get(
                "/handlers",
                params={
                    "status": status,
                    "workflow_name": workflow_name,
                },
            )
            _raise_for_status_with_body(response)

            return HandlersListResponse.model_validate(response.json())

    async def get_handler(self, handler_id: str) -> HandlerData:
        """Get a workflow handler by ID.

        Returns handler metadata including status, result (if completed),
        and timestamps.

        Args:
            handler_id: ID of the handler.
        """
        async with self._get_client() as client:
            response = await client.get(f"/handlers/{handler_id}")
            _raise_for_status_with_body(response)

            return HandlerData.model_validate(response.json())

    async def wait_for_handler(
        self,
        handler_id: str,
        *,
        timeout: float | None = None,
        poll_interval: float = 0.5,
        max_poll_interval: float = 5.0,
        backoff: Literal["constant", "linear", "exponential"] = "linear",
    ) -> HandlerData:
        """Wait until a workflow handler reaches a terminal status.

        Args:
            handler_id: ID of the handler to poll.
            timeout: Maximum number of seconds to wait. ``None`` waits
                indefinitely.
            poll_interval: Initial number of seconds between status requests.
            max_poll_interval: Maximum number of seconds between requests.
            backoff: Strategy for increasing the polling interval after each
                non-terminal response.

        Raises:
            TimeoutError: If the handler is still running after ``timeout``.
            ValueError: If ``timeout`` or ``poll_interval`` is negative.
            httpx.HTTPStatusError: If a status request fails.
        """
        if timeout is not None and timeout < 0:
            raise ValueError("timeout must be non-negative or None")
        if poll_interval < 0:
            raise ValueError("poll_interval must be non-negative")
        if max_poll_interval < 0:
            raise ValueError("max_poll_interval must be non-negative")
        if max_poll_interval < poll_interval:
            raise ValueError(
                "max_poll_interval must be greater than or equal to poll_interval"
            )
        if backoff not in ("constant", "linear", "exponential"):
            raise ValueError(f"invalid backoff strategy: {backoff}")

        loop = asyncio.get_running_loop()
        deadline = None if timeout is None else loop.time() + timeout
        current_interval = poll_interval

        while True:
            handler = await self.get_handler(handler_id)
            if is_status_completed(handler.status):
                return handler

            if deadline is None:
                sleep_interval = current_interval
            else:
                remaining = deadline - loop.time()
                if remaining <= 0:
                    raise TimeoutError(
                        f"Handler {handler_id!r} did not complete within {timeout} seconds"
                    )
                sleep_interval = min(current_interval, remaining)

            await asyncio.sleep(sleep_interval)
            current_interval = _next_poll_interval(
                current_interval, backoff, max_poll_interval
            )

    async def cancel_handler(
        self, handler_id: str, purge: bool = False
    ) -> CancelHandlerResponse:
        """Cancel a running workflow.

        Args:
            handler_id: ID of the handler to cancel.
            purge: Also remove the handler from the persistence store.
                Defaults to ``False``.
        """
        async with self._get_client() as client:
            response = await client.post(
                f"/handlers/{handler_id}/cancel",
                params={"purge": "true" if purge else "false"},
            )
            _raise_for_status_with_body(response)

            return CancelHandlerResponse.model_validate(response.json())

is_healthy async

is_healthy() -> HealthResponse

Check whether the workflow server is healthy.

Raises:

Type	Description
HTTPStatusError	If the server returns an error status.

Source code in llama_agents/client/client.py

async def is_healthy(self) -> HealthResponse:
    """Check whether the workflow server is healthy.

    Raises:
        httpx.HTTPStatusError: If the server returns an error status.
    """
    async with self._get_client() as client:
        response = await client.get("/health")
        _raise_for_status_with_body(response)
        return HealthResponse.model_validate(response.json())

list_workflows async

list_workflows() -> WorkflowsListResponse

List the names of all workflows registered on the server.

Source code in llama_agents/client/client.py

async def list_workflows(self) -> WorkflowsListResponse:
    """List the names of all workflows registered on the server."""
    async with self._get_client() as client:
        response = await client.get("/workflows")

        _raise_for_status_with_body(response)

        return WorkflowsListResponse.model_validate(response.json())

get_workflow_schema async

get_workflow_schema(workflow_name: str) -> WorkflowSchemaResponse

Get the start and stop event schemas for a workflow.

Source code in llama_agents/client/client.py

async def get_workflow_schema(self, workflow_name: str) -> WorkflowSchemaResponse:
    """Get the start and stop event schemas for a workflow."""
    async with self._get_client() as client:
        response = await client.get(f"/workflows/{workflow_name}/schema")
        _raise_for_status_with_body(response)
        return WorkflowSchemaResponse.model_validate(response.json())

get_workflow_events_schema async

get_workflow_events_schema(workflow_name: str) -> WorkflowEventsListResponse

Get the schemas of events registered for a workflow.

Source code in llama_agents/client/client.py

async def get_workflow_events_schema(
    self, workflow_name: str
) -> WorkflowEventsListResponse:
    """Get the schemas of events registered for a workflow."""
    async with self._get_client() as client:
        response = await client.get(f"/workflows/{workflow_name}/events")
        _raise_for_status_with_body(response)
        return WorkflowEventsListResponse.model_validate(response.json())

get_workflow_graph async

get_workflow_graph(workflow_name: str) -> WorkflowGraphResponse

Get the directed graph representation of a workflow.

Source code in llama_agents/client/client.py

async def get_workflow_graph(self, workflow_name: str) -> WorkflowGraphResponse:
    """Get the directed graph representation of a workflow."""
    async with self._get_client() as client:
        response = await client.get(f"/workflows/{workflow_name}/representation")
        _raise_for_status_with_body(response)
        return WorkflowGraphResponse.model_validate(response.json())

run_workflow async

run_workflow(workflow_name: str, handler_id: str | None = None, start_event: StartEvent | dict[str, Any] | None = None, context: Context | dict[str, Any] | None = None) -> HandlerData

Run the workflow and block until completion.

Parameters:

Name	Type	Description	Default
workflow_name	str	Name of the registered workflow to run.	required
start_event	StartEvent | dict[str, Any] | None	Input event for the workflow. Can be a StartEvent instance or a plain dict.	None
context	Context | dict[str, Any] | None	Workflow context to restore, for continuing a previous run.	None
handler_id	str | None	Handler identifier to continue from a previous completed run.	None

Returns:

Name	Type	Description
HandlerData	HandlerData	Handler metadata including the final result.

Source code in llama_agents/client/client.py

async def run_workflow(
    self,
    workflow_name: str,
    handler_id: str | None = None,
    start_event: StartEvent | dict[str, Any] | None = None,
    context: Context | dict[str, Any] | None = None,
) -> HandlerData:
    """Run the workflow and block until completion.

    Args:
        workflow_name: Name of the registered workflow to run.
        start_event: Input event for the workflow. Can be a ``StartEvent``
            instance or a plain dict.
        context: Workflow context to restore, for continuing a previous run.
        handler_id: Handler identifier to continue from a previous
            completed run.

    Returns:
        HandlerData: Handler metadata including the final result.
    """
    if start_event is not None:
        try:
            start_event = _serialize_event(start_event, bare=True)
        except Exception as e:
            raise ValueError(
                f"Impossible to serialize the start event because of: {e}"
            )
    if isinstance(context, Context):
        try:
            context = context.to_dict()
        except Exception as e:
            raise ValueError(f"Impossible to serialize the context because of: {e}")
    request_body: dict[str, Any] = {
        "start_event": start_event
        if start_event is not None
        else _serialize_event(StartEvent(), bare=True),
        "context": context if context is not None else {},
    }
    if handler_id:
        request_body["handler_id"] = handler_id
    async with self._get_client() as client:
        response = await client.post(
            f"/workflows/{workflow_name}/run", json=request_body
        )

        _raise_for_status_with_body(response)

        return HandlerData.model_validate(response.json())

run_workflow_nowait async

run_workflow_nowait(workflow_name: str, handler_id: str | None = None, start_event: StartEvent | dict[str, Any] | None = None, context: Context | dict[str, Any] | None = None) -> HandlerData

Start the workflow without waiting for completion.

Use the returned handler_id to stream events, poll for results, or send events.

Parameters:

Name	Type	Description	Default
workflow_name	str	Name of the registered workflow to run.	required
start_event	StartEvent | dict[str, Any] | None	Input event for the workflow. Can be a StartEvent instance or a plain dict.	None
context	Context | dict[str, Any] | None	Workflow context to restore, for continuing a previous run.	None
handler_id	str | None	Handler identifier to continue from a previous completed run.	None

Returns:

Name	Type	Description
HandlerData	HandlerData	Handler metadata including the handler_id.

Source code in llama_agents/client/client.py

async def run_workflow_nowait(
    self,
    workflow_name: str,
    handler_id: str | None = None,
    start_event: StartEvent | dict[str, Any] | None = None,
    context: Context | dict[str, Any] | None = None,
) -> HandlerData:
    """Start the workflow without waiting for completion.

    Use the returned ``handler_id`` to stream events, poll for results,
    or send events.

    Args:
        workflow_name: Name of the registered workflow to run.
        start_event: Input event for the workflow. Can be a ``StartEvent``
            instance or a plain dict.
        context: Workflow context to restore, for continuing a previous run.
        handler_id: Handler identifier to continue from a previous
            completed run.

    Returns:
        HandlerData: Handler metadata including the ``handler_id``.
    """
    if start_event is not None:
        try:
            start_event = _serialize_event(start_event)
        except Exception as e:
            raise ValueError(
                f"Impossible to serialize the start event because of: {e}"
            )
    if isinstance(context, Context):
        try:
            context = context.to_dict()
        except Exception as e:
            raise ValueError(f"Impossible to serialize the context because of: {e}")
    request_body: dict[str, Any] = {
        "start_event": start_event
        if start_event is not None
        else _serialize_event(StartEvent()),
        "context": context if context is not None else {},
    }
    if handler_id:
        request_body["handler_id"] = handler_id
    async with self._get_client() as client:
        response = await client.post(
            f"/workflows/{workflow_name}/run-nowait", json=request_body
        )

        _raise_for_status_with_body(response)

        return HandlerData.model_validate(response.json())

get_workflow_events

get_workflow_events(handler_id: str, include_internal_events: bool = False, after_sequence: int | Literal['now'] = -1, max_reconnect_attempts: int = 3) -> EventStream

Stream events as they are produced by the workflow.

Returns an EventStream whose last_sequence property tracks the sequence number of the most recently yielded event. Uses SSE and automatically reconnects from the last received event on connection drops.

Example:

stream = client.get_workflow_events(handler_id)
async for event in stream:
    print(event.type, stream.last_sequence)

Parameters:

Name	Type	Description	Default
handler_id	str	ID of the handler running the workflow.	required
include_internal_events	bool	Include internal dispatch events. Defaults to False.	False
after_sequence	int | Literal['now']	Where to start streaming. -1 (default) streams all events from the beginning. "now" skips existing events and only delivers new ones. An integer N streams events after sequence N.	-1
max_reconnect_attempts	int	Maximum reconnect attempts on connection drop. Defaults to 3.	3

Source code in llama_agents/client/client.py

def get_workflow_events(
    self,
    handler_id: str,
    include_internal_events: bool = False,
    after_sequence: int | Literal["now"] = -1,
    max_reconnect_attempts: int = 3,
) -> EventStream:
    """Stream events as they are produced by the workflow.

    Returns an ``EventStream`` whose ``last_sequence`` property tracks
    the sequence number of the most recently yielded event. Uses SSE
    and automatically reconnects from the last received event on
    connection drops.

    Example:

        stream = client.get_workflow_events(handler_id)
        async for event in stream:
            print(event.type, stream.last_sequence)

    Args:
        handler_id: ID of the handler running the workflow.
        include_internal_events: Include internal dispatch events.
            Defaults to ``False``.
        after_sequence: Where to start streaming. ``-1`` (default) streams
            all events from the beginning. ``"now"`` skips existing events
            and only delivers new ones. An integer ``N`` streams events
            after sequence ``N``.
        max_reconnect_attempts: Maximum reconnect attempts on connection
            drop. Defaults to ``3``.
    """
    queue: asyncio.Queue[_QueueItem] = asyncio.Queue()
    stream = EventStream(queue, None, after_sequence)

    async def reader() -> None:
        incl_inter = "true" if include_internal_events else "false"
        url = f"/events/{handler_id}"
        last_sequence: int | Literal["now"] = after_sequence
        attempts = 0
        try:
            while True:
                async with self._get_client() as client:
                    try:
                        async with client.stream(
                            "GET",
                            url,
                            params={
                                "sse": "true",
                                "include_internal": incl_inter,
                                "after_sequence": str(last_sequence),
                            },
                            headers={"Connection": "keep-alive"},
                            timeout=None,
                        ) as response:
                            if response.status_code == 404:
                                raise ValueError("Handler not found")
                            elif response.status_code == 204:
                                await queue.put(_QueuedDone())
                                return

                            _raise_for_status_with_body(response)

                            # Reset attempts on successful connection
                            attempts = 0

                            # Parse SSE stream: "id: N\ndata: {...}\n\n"
                            current_id: str | None = None
                            async for line in response.aiter_lines():
                                stripped = line.strip()
                                if not stripped:
                                    # Empty line = end of SSE event
                                    continue
                                if stripped.startswith("id:"):
                                    current_id = stripped[3:].strip()
                                elif stripped.startswith("data:"):
                                    data = stripped[5:].strip()
                                    event = EventEnvelopeWithMetadata.model_validate_json(
                                        data
                                    )
                                    if current_id is not None:
                                        try:
                                            last_sequence = int(current_id)
                                        except ValueError:
                                            pass
                                    await queue.put(
                                        _QueuedEvent(
                                            sequence=last_sequence,
                                            event=event,
                                        )
                                    )
                                    current_id = None

                        # Stream ended normally (server closed connection)
                        await queue.put(_QueuedDone())
                        return

                    except httpx.TimeoutException:
                        raise TimeoutError(
                            f"Timeout waiting for events from handler {handler_id}"
                        )
                    except (httpx.RequestError, ConnectionError):
                        attempts += 1
                        if attempts > max_reconnect_attempts:
                            raise ConnectionError(
                                f"Failed to connect to event stream after {max_reconnect_attempts} attempts"
                            )
                        # Retry from last received sequence
        except asyncio.CancelledError:
            await queue.put(_QueuedDone())
        except BaseException as exc:
            await queue.put(_QueuedError(exc))

    stream._task = asyncio.create_task(reader())
    return stream

send_event async

send_event(handler_id: str, event: Event | dict[str, Any], step: str | None = None) -> SendEventResponse

Send an event to a running workflow.

Useful for human-in-the-loop workflows that wait for external input.

Parameters:

Name	Type	Description	Default
handler_id	str	ID of the handler running the workflow.	required
event	Event | dict[str, Any]	Event to send, as an Event instance or a dict.	required
step	str | None	Target a specific workflow step. When None, the event is broadcast to all waiting steps.	None

Source code in llama_agents/client/client.py

async def send_event(
    self,
    handler_id: str,
    event: Event | dict[str, Any],
    step: str | None = None,
) -> SendEventResponse:
    """Send an event to a running workflow.

    Useful for human-in-the-loop workflows that wait for external input.

    Args:
        handler_id: ID of the handler running the workflow.
        event: Event to send, as an ``Event`` instance or a dict.
        step: Target a specific workflow step. When ``None``, the event
            is broadcast to all waiting steps.
    """
    try:
        serialized_event: dict[str, Any] = _serialize_event(event)
    except Exception as e:
        raise ValueError(f"Error while serializing the provided event: {e}")
    request_body: dict[str, Any] = {"event": serialized_event}
    if step:
        request_body["step"] = step
    async with self._get_client() as client:
        response = await client.post(f"/events/{handler_id}", json=request_body)
        _raise_for_status_with_body(response)

        return SendEventResponse.model_validate(response.json())

get_handlers async

get_handlers(status: list[Status] | None = None, workflow_name: list[str] | None = None) -> HandlersListResponse

List all workflow handlers.

Parameters:

Name	Type	Description	Default
status	list[Status] | None	Filter by handler status (e.g. "running", "completed").	None
workflow_name	list[str] | None	Filter by workflow name.	None

Source code in llama_agents/client/client.py

async def get_handlers(
    self,
    status: list[Status] | None = None,
    workflow_name: list[str] | None = None,
) -> HandlersListResponse:
    """List all workflow handlers.

    Args:
        status: Filter by handler status (e.g. ``"running"``,
            ``"completed"``).
        workflow_name: Filter by workflow name.
    """
    async with self._get_client() as client:
        response = await client.get(
            "/handlers",
            params={
                "status": status,
                "workflow_name": workflow_name,
            },
        )
        _raise_for_status_with_body(response)

        return HandlersListResponse.model_validate(response.json())

get_handler async

get_handler(handler_id: str) -> HandlerData

Get a workflow handler by ID.

Returns handler metadata including status, result (if completed), and timestamps.

Parameters:

Name	Type	Description	Default
handler_id	str	ID of the handler.	required

Source code in llama_agents/client/client.py

async def get_handler(self, handler_id: str) -> HandlerData:
    """Get a workflow handler by ID.

    Returns handler metadata including status, result (if completed),
    and timestamps.

    Args:
        handler_id: ID of the handler.
    """
    async with self._get_client() as client:
        response = await client.get(f"/handlers/{handler_id}")
        _raise_for_status_with_body(response)

        return HandlerData.model_validate(response.json())

wait_for_handler async

wait_for_handler(handler_id: str, *, timeout: float | None = None, poll_interval: float = 0.5, max_poll_interval: float = 5.0, backoff: Literal['constant', 'linear', 'exponential'] = 'linear') -> HandlerData

Wait until a workflow handler reaches a terminal status.

Parameters:

Name	Type	Description	Default
handler_id	str	ID of the handler to poll.	required
timeout	float | None	Maximum number of seconds to wait. None waits indefinitely.	None
poll_interval	float	Initial number of seconds between status requests.	0.5
max_poll_interval	float	Maximum number of seconds between requests.	5.0
backoff	Literal['constant', 'linear', 'exponential']	Strategy for increasing the polling interval after each non-terminal response.	'linear'

Raises:

Type	Description
TimeoutError	If the handler is still running after timeout.
ValueError	If timeout or poll_interval is negative.
HTTPStatusError	If a status request fails.

Source code in llama_agents/client/client.py

async def wait_for_handler(
    self,
    handler_id: str,
    *,
    timeout: float | None = None,
    poll_interval: float = 0.5,
    max_poll_interval: float = 5.0,
    backoff: Literal["constant", "linear", "exponential"] = "linear",
) -> HandlerData:
    """Wait until a workflow handler reaches a terminal status.

    Args:
        handler_id: ID of the handler to poll.
        timeout: Maximum number of seconds to wait. ``None`` waits
            indefinitely.
        poll_interval: Initial number of seconds between status requests.
        max_poll_interval: Maximum number of seconds between requests.
        backoff: Strategy for increasing the polling interval after each
            non-terminal response.

    Raises:
        TimeoutError: If the handler is still running after ``timeout``.
        ValueError: If ``timeout`` or ``poll_interval`` is negative.
        httpx.HTTPStatusError: If a status request fails.
    """
    if timeout is not None and timeout < 0:
        raise ValueError("timeout must be non-negative or None")
    if poll_interval < 0:
        raise ValueError("poll_interval must be non-negative")
    if max_poll_interval < 0:
        raise ValueError("max_poll_interval must be non-negative")
    if max_poll_interval < poll_interval:
        raise ValueError(
            "max_poll_interval must be greater than or equal to poll_interval"
        )
    if backoff not in ("constant", "linear", "exponential"):
        raise ValueError(f"invalid backoff strategy: {backoff}")

    loop = asyncio.get_running_loop()
    deadline = None if timeout is None else loop.time() + timeout
    current_interval = poll_interval

    while True:
        handler = await self.get_handler(handler_id)
        if is_status_completed(handler.status):
            return handler

        if deadline is None:
            sleep_interval = current_interval
        else:
            remaining = deadline - loop.time()
            if remaining <= 0:
                raise TimeoutError(
                    f"Handler {handler_id!r} did not complete within {timeout} seconds"
                )
            sleep_interval = min(current_interval, remaining)

        await asyncio.sleep(sleep_interval)
        current_interval = _next_poll_interval(
            current_interval, backoff, max_poll_interval
        )

cancel_handler async

cancel_handler(handler_id: str, purge: bool = False) -> CancelHandlerResponse

Cancel a running workflow.

Parameters:

Name	Type	Description	Default
handler_id	str	ID of the handler to cancel.	required
purge	bool	Also remove the handler from the persistence store. Defaults to False.	False

Source code in llama_agents/client/client.py

async def cancel_handler(
    self, handler_id: str, purge: bool = False
) -> CancelHandlerResponse:
    """Cancel a running workflow.

    Args:
        handler_id: ID of the handler to cancel.
        purge: Also remove the handler from the persistence store.
            Defaults to ``False``.
    """
    async with self._get_client() as client:
        response = await client.post(
            f"/handlers/{handler_id}/cancel",
            params={"purge": "true" if purge else "false"},
        )
        _raise_for_status_with_body(response)

        return CancelHandlerResponse.model_validate(response.json())

EventStream

Async iterator over workflow events that exposes the current stream position.

Returned by WorkflowClient.get_workflow_events(). Use last_sequence to capture the cursor for resuming later::

stream = client.get_workflow_events(handler_id)
async for event in stream:
    print(event.type, stream.last_sequence)

# Resume from where we left off:
stream = client.get_workflow_events(
    handler_id, after_sequence=stream.last_sequence
)

Source code in llama_agents/client/client.py

class EventStream:
    """Async iterator over workflow events that exposes the current stream position.

    Returned by ``WorkflowClient.get_workflow_events()``. Use
    ``last_sequence`` to capture the cursor for resuming later::

        stream = client.get_workflow_events(handler_id)
        async for event in stream:
            print(event.type, stream.last_sequence)

        # Resume from where we left off:
        stream = client.get_workflow_events(
            handler_id, after_sequence=stream.last_sequence
        )
    """

    def __init__(
        self,
        queue: asyncio.Queue[_QueueItem],
        task: asyncio.Task[None] | None,
        initial_sequence: int | Literal["now"],
    ) -> None:
        self._queue = queue
        self._task = task
        self._last_sequence: int | Literal["now"] = initial_sequence
        self._iter_started = False

    @property
    def last_sequence(self) -> int | Literal["now"]:
        """The sequence number of the most recently yielded event, or the
        initial ``after_sequence`` value if no events have been yielded yet."""
        return self._last_sequence

    def __aiter__(self) -> AsyncIterator[EventEnvelopeWithMetadata]:
        if self._iter_started:
            raise RuntimeError("EventStream can only be iterated once")
        self._iter_started = True
        return self._iterate()

    async def _iterate(self) -> AsyncGenerator[EventEnvelopeWithMetadata, None]:
        try:
            while True:
                item = await self._queue.get()
                if isinstance(item, _QueuedDone):
                    return
                if isinstance(item, _QueuedError):
                    raise item.error
                self._last_sequence = item.sequence
                yield item.event
        finally:
            await self.aclose()

    async def aclose(self) -> None:
        """Cancel the background reader and release resources."""
        if self._task is None:
            return
        task, self._task = self._task, None
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):
            pass

last_sequence property

last_sequence: int | Literal['now']

The sequence number of the most recently yielded event, or the initial after_sequence value if no events have been yielded yet.

aclose async

aclose() -> None

Cancel the background reader and release resources.

Source code in llama_agents/client/client.py

async def aclose(self) -> None:
    """Cancel the background reader and release resources."""
    if self._task is None:
        return
    task, self._task = self._task, None
    task.cancel()
    try:
        await task
    except (asyncio.CancelledError, Exception):
        pass