WorkflowClient 类提供了一个 Python 接口，用于与正在运行的 WorkflowServer 交互。它支持列出工作流、以同步或异步方式运行、流式接收事件，以及为人在环路的工作流发送事件。

安装

该客户端是独立于核心 llama-index-workflows 库的包：

终端窗口

pip install llama-agents-client

设置

from llama_agents.client import WorkflowClient

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

你也可以传入一个预先配置好的 httpx.AsyncClient，而不是 base_url：

import httpx

httpx_client = httpx.AsyncClient(base_url="http://0.0.0.0:8080", headers={"Authorization": "Bearer ..."})
client = WorkflowClient(httpx_client=httpx_client)

基本用法

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

async def main():
    client = WorkflowClient(base_url="http://0.0.0.0:8080")

    # Check server health
    await client.is_healthy()

    # List available workflows
    workflows = await client.list_workflows()
    print(workflows)

    # Run a workflow synchronously (blocks until completion)
    result = await client.run_workflow("greet", start_event=StartEvent(name="John"))
    print(result.result)

异步运行与事件流

对于长时间运行的工作流，可以异步启动它，并在事件产生时就流式接收：

handler = await client.run_workflow_nowait("greet", start_event=StartEvent(name="John"))
handler_id = handler.handler_id

async for event in client.get_workflow_events(handler_id):
    print("Received:", event.type, event.value)

result = await client.get_handler(handler_id)
print(f"Final result: {result.result}")

游标行为（after_sequence）

get_workflow_events 接受一个 after_sequence 参数，控制从事件历史的哪个位置开始推送：

取值	行为
-1（默认）	从最开始流式推送全部事件，包括你开始监听之前就已产生的那些。
"now"	跳过所有已有事件，只推送请求到达之后产生的事件。
任意整数 N	推送序号大于 N 的事件。

这几种取值分别适合不同场景：

• 完整回放（-1）：你想看到一次运行的全部事件，即使它已经在进行中或已经完成。
• 实时跟随（"now"）：工作流由你自己启动，你只关心往后新增的事件。如果工作流已经完成、所有事件都已产生，服务端会返回 HTTP 204，流立即结束。
• 从断点续传（整数）：你之前断开了连接，想从最后看到的那个事件继续。

要拿到流的位置，用 get_workflow_events，它返回一个带 last_sequence 属性的 EventStream 对象：

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

# Save the position for later
saved_sequence = stream.last_sequence

随后就可以从保存过的位置续传：

stream = client.get_workflow_events(handler_id, after_sequence=saved_sequence)
async for event in stream:
    print(event)

或者跳过已有事件，只接收新事件：

stream = client.get_workflow_events(handler_id, after_sequence="now")
async for event in stream:
    print(event)

连接中断时，get_workflow_events 会自动从最后收到的序号重连（最多 max_reconnect_attempts 次，默认 3）。

人在环路

对于需要外部输入的工作流，可以把事件流与 send_event 结合使用：

from workflows import Workflow, step
from workflows.context import Context
from workflows.events import (
    StartEvent,
    StopEvent,
    InputRequiredEvent,
    HumanResponseEvent,
)
from llama_agents.server import WorkflowServer

class RequestEvent(InputRequiredEvent):
    prompt: str

class ResponseEvent(HumanResponseEvent):
    response: str

class OutEvent(StopEvent):
    output: str

class HumanInTheLoopWorkflow(Workflow):
    @step
    async def prompt_human(self, ev: StartEvent, ctx: Context) -> RequestEvent:
        return RequestEvent(prompt="What is your name?")

    @step
    async def greet_human(self, ev: ResponseEvent) -> OutEvent:
        return OutEvent(output=f"Hello, {ev.response}")

server = WorkflowServer()
server.add_workflow("human", HumanInTheLoopWorkflow(timeout=1000))
await server.serve("0.0.0.0", "8080")

然后在客户端侧：

from llama_agents.client import WorkflowClient

client = WorkflowClient(base_url="http://0.0.0.0:8080")
handler = await client.run_workflow_nowait("human")
handler_id = handler.handler_id

async for event in client.get_workflow_events(handler_id):
    # load_event() reconstructs the typed Event class by qualified name
    loaded = event.load_event()
    if isinstance(loaded, RequestEvent):
        print("Workflow is requiring human input:", loaded.prompt)
        name = input("Reply here: ")
        await client.send_event(
            handler_id=handler_id,
            event=ResponseEvent(response=name.capitalize().strip()),
        )

result = await client.get_handler(handler_id)
res = OutEvent.model_validate(result.result)
print("Received final message:", res.output)

当事件类可以通过其限定名导入时，load_event() 会自动生效。传入 serializer=JsonSerializer(allowed_types=[RequestEvent, ResponseEvent]) 可以限制客户端能解析哪些类，或把这些类作为 registry 列表传入。如果你不需要带类型的事件，原始的 event.type 与 event.value 字典始终可用。