如何从Python Temporal工作流中调用JavaScript活动？

1. 在Python中定义跨语言活动接口 #

先在Python代码里定义活动契约，确保和JavaScript实现的活动名称、参数/返回值类型匹配：

from temporalio import activity

@activity.defn(name="your-javascript-activity")  # 该名称需与JS侧活动名称完全一致
async def your_javascript_activity(name: str) -> str:
    # 仅定义接口，实际逻辑在JavaScript中实现
    raise NotImplementedError("Activity implemented in JavaScript")

2. 在JavaScript中实现活动逻辑 #

使用Temporal JavaScript SDK编写活动代码，并注册到Worker：

import { Worker } from '@temporalio/worker';
import { Connection } from '@temporalio/client';

// 实现活动函数，名称需与Python侧定义的name一致
async function yourJavascriptActivity(name) {
  // 这里编写你的业务逻辑
  return `Hello from JavaScript, ${name}!`;
}

async function runWorker() {
  const connection = await Connection.connect();
  const worker = await Worker.create({
    connection,
    taskQueue: 'shared-task-queue',  // 与Python工作流使用同一个任务队列
    activities: {
      'your-javascript-activity': yourJavascriptActivity,  // 映射活动名称到具体实现
    },
  });

  await worker.run();
}

runWorker().catch(err => console.error(err));

3. 修改Python工作流调用活动 #

更新你的Python工作流代码，调用定义好的活动：

from temporalio import workflow
from temporalio.workflow import execute_activity

@workflow.defn
class YourSchedulesWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        # 调用跨语言活动，指定活动函数、参数、任务队列和超时时间
        return await execute_activity(
            your_javascript_activity,
            name,
            task_queue="shared-task-queue",
            start_to_close_timeout=workflow.duration(minutes=5),
        )

4. 启动Worker并运行工作流 #

• 启动JavaScript活动Worker：执行上述JS代码（例如node worker.js）
• 启动Python工作流Worker，确保监听同一个任务队列：

import asyncio
from temporalio import worker
from temporalio.client import Client

async def run_worker():
    client = await Client.connect("localhost:7233")
    w = worker.Worker(
        client,
        task_queue="shared-task-queue",
        workflows=[YourSchedulesWorkflow],
        activities=[],  # Python侧无需注册该活动，由JS Worker处理
    )
    await w.run()

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

关键注意事项 #

• 活动名称匹配：Python中@activity.defn(name="xxx")的名称必须与JS Worker注册的活动名称完全一致
• 任务队列一致：Python工作流、Python工作流Worker、JS活动Worker必须使用同一个任务队列
• 序列化兼容：活动的参数和返回值需为JSON可序列化类型（如字符串、数字、列表、字典），Temporal会自动处理跨语言序列化

内容的提问来源于stack exchange，提问作者Vasanth Kumar