run_worker.py

"""Worker for the hello world sample (Functional API)."""

import asyncio
import os

from temporalio.client import Client
from temporalio.contrib.langgraph import LangGraphPlugin
from temporalio.worker import Worker

from langgraph_plugin.functional_api.hello_world.workflow import (
    HelloWorldFunctionalWorkflow,
    activity_options,
    all_tasks,
    hello_entrypoint,
)


async def main() -> None:
    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
    plugin = LangGraphPlugin(
        entrypoints={"hello-world": hello_entrypoint},
        tasks=all_tasks,
        activity_options=activity_options,
    )

    worker = Worker(
        client,
        task_queue="langgraph-hello-world-functional",
        workflows=[HelloWorldFunctionalWorkflow],
        plugins=[plugin],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()


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