Spaces:
Configuration error
Configuration error
| """LangGraph Platform client - run/stream agent graphs, thread state.""" | |
| import os | |
| import requests | |
| BASE = os.getenv("LANGGRAPH_API_URL", "http://localhost:8123") | |
| TOKEN = os.getenv("LANGGRAPH_AUTH_TOKEN") | |
| def _h(): | |
| headers = {"Content-Type": "application/json"} | |
| if TOKEN: | |
| headers["Authorization"] = f"Bearer {TOKEN}" | |
| return headers | |
| def invoke(graph_id, input, thread_id=None, config=None): | |
| body = {"input": input, "config": config or {}} | |
| if thread_id: | |
| body["config"] = {"configurable": {"thread_id": thread_id}} | |
| return requests.post(f"{BASE}/{graph_id}/invoke", headers=_h(), json=body, timeout=120).json() | |
| def stream(graph_id, input, thread_id=None): | |
| body = {"input": input, "config": {"configurable": {"thread_id": thread_id}} if thread_id else {}} | |
| r = requests.post(f"{BASE}/{graph_id}/stream", headers=_h(), json=body, timeout=300, stream=True) | |
| r.raise_for_status() | |
| for line in r.iter_lines(decode_unicode=True): | |
| if line: | |
| yield line | |
| def get_state(graph_id, thread_id): | |
| return requests.get(f"{BASE}/{graph_id}/state", headers=_h(), params={"thread_id": thread_id}, timeout=60).json() | |
| def list_threads(graph_id, **params): | |
| return requests.get(f"{BASE}/{graph_id}/threads", headers=_h(), params=params, timeout=60).json() | |