forked from temporalio/samples-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrun_worker.py
More file actions
31 lines (23 loc) · 779 Bytes
/
Copy pathrun_worker.py
File metadata and controls
31 lines (23 loc) · 779 Bytes
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
"""Worker for the ReAct agent (Graph API)."""
import asyncio
import os
from temporalio.client import Client
from temporalio.contrib.langgraph import LangGraphPlugin
from temporalio.worker import Worker
from langgraph_plugin.graph_api.react_agent.workflow import (
ReactAgentWorkflow,
make_agent_graph,
)
async def main() -> None:
client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
plugin = LangGraphPlugin(graphs={"react-agent": make_agent_graph()})
worker = Worker(
client,
task_queue="langgraph-react-agent",
workflows=[ReactAgentWorkflow],
plugins=[plugin],
)
print("Worker started. Ctrl+C to exit.")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())