Spaces:
Configuration error
Configuration error
File size: 1,317 Bytes
aa682d9 | 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 32 33 34 35 36 37 38 | """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()
|