Temporal 客户端可以与 Temporal 服务端进行通信。与 Temporal 服务端通信可以执行各种操作，比如启动工作流执行、向工作流执行发送信号和查询、获取工作流结果等。

对于独立 Activity，Temporal 客户端还可以直接启动和管理独立 Activity，而无需引入工作流。

不能在工作流内部初始化和使用 Temporal 客户端。但在 Activity 内部使用 Temporal 客户端与 Temporal 服务端通信很常见。

连接到开发 Temporal 服务端

使用 Client.connect 创建客户端。连接选项包括 Temporal 服务端地址、命名空间（Namespace）以及可选的 TLS 配置。可以直接在代码中提供这些选项，也可以通过环境变量或使用 envconfig 辅助工具从 TOML 配置文件中加载。为安全及复用配置考虑，推荐使用环境变量或配置文件。

当在本地运行 Temporal 服务端时（比如使用 Temporal CLI 开发服务），所需的配置项极少。如果未指定主机/端口，大多数连接默认使用 127.0.0.1:7233 和 default 命名空间。

如果不想使用环境变量或配置文件，也可以直接在代码中指定连接选项。这在本地开发和测试时非常方便。也可以先从环境变量或配置文件中加载基础配置，然后在代码中覆盖特定选项。

使用 Client 类上的 connect() 方法，创建 Temporal 客户端，连接到 Temporal 服务。

import asyncio
from temporalio.client import Client
from your_workflows import YourWorkflow


async def main():
    client =await Client.connect("localhost:7233")

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

启动工作流执行

工作流执行语义依赖多个参数——要启动工作流执行，必须提供任务队列（且该队列有 Worker 在轮询）、工作流类型、语言特定的上下文数据以及工作流函数的参数。

在下面的示例中，所有工作流执行均使用 Temporal 客户端启动。要从另一个工作流执行中启动工作流执行，请使用子工作流（Child Workflow）或外部工作流（External Workflow）API。

启动工作流执行的请求使 Temporal 服务端在工作流执行事件历史中创建第一个事件（WorkflowExecutionStarted）。随后，Temporal 服务端创建第一个工作流任务，从而产生第一个 WorkflowTaskScheduled 事件。

要在 Python 中启动工作流执行，需使用客户端的 start_workflow() 或 execute_workflow() 异步方法。

# ...
async def main():
    client = await Client.connect("localhost:7233")

    result =await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

设置工作流的任务队列

在大多数 SDK 中，唯一必须设置的工作流选项是任务队列的名称。

任何代码要执行，都必须有 Worker 进程在运行，该进程中包含 Worker 实体，并且该 Worker 实体轮询相同的任务队列名称。

要在 Python 中设置任务队列，需要在调用 start_workflow() 或 execute_workflow() 方法执行工作流时指定 task_queue 参数。

# ...
async def main():
    client = await Client.connect("localhost:7233")

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

设置工作流 ID

必须设置工作流 ID（Workflow Id）。

在设置工作流 ID 时，建议将其映射到业务流程或业务实体标识符，比如订单标识符或客户标识符。

要在 Python 中设置工作流 ID，需要在调用 start_workflow() 或 execute_workflow() 方法执行工作流时指定 id 参数。

id 参数应为工作流执行的唯一标识符。

# ...
async def main():
    client = await Client.connect("localhost:7233")

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

获取工作流执行的结果

如果启动工作流执行的调用成功，那么将获得该工作流执行的运行 ID（Run Id）。

工作流 ID、运行 ID 和命名空间可用于在系统中唯一标识工作流执行，及获取其结果。

既可以阻塞等待结果（同步执行），也可以在其他时间点获取结果（异步执行）。

在 Temporal 平台中，也可以使用查询（Query）作为获取工作流执行状态和结果的方式。

使用 start_workflow() 或 get_workflow_handle() 返回工作流句柄，然后使用 result 方法等待工作流的结果。

要通过工作流 ID 获取现有工作流的句柄，可以使用 get_workflow_handle()，或者使用 get_workflow_handle_for()。

然后使用 describe() 获取工作流的当前状态。如果工作流不存在，此调用将失败。

import asyncio
from temporalio.client import Client


async def main():
    client = await Client.connect("localhost:7233")

    handle = client.get_workflow_handle(
        workflow_id="your-workflow-id",
    )
    results =await handle.result()
    print(f"Result: {results}")


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