Access LangGraph store from inside a graph node or entrypoint task at runtime.
Can be called from inside any StateGraph node or
functional API task, as long as the StateGraph or the entrypoint
was initialized with a store, e.g.:
# with StateGraph
graph = (
StateGraph(...)
...
.compile(store=store)
)
# or with entrypoint
@entrypoint(store=store)
def workflow(inputs):
...If you are using Python < 3.11 and are running LangGraph asynchronously,
get_store() won't work since it uses contextvar propagation (only available in Python >= 3.11).
Using with StateGraph:
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.store.memory import InMemoryStore
from langgraph.config import get_store
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
class State(TypedDict):
foo: int
def my_node(state: State):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return {"foo": stored_value + 1}
graph = (
StateGraph(State)
.add_node(my_node)
.add_edge(START, "my_node")
.compile(store=store)
)
graph.invoke({"foo": 1}){"foo": 3}
Using with functional API:
from langgraph.func import entrypoint, task
from langgraph.store.memory import InMemoryStore
from langgraph.config import get_store
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
@task
def my_task(value: int):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return stored_value + 1
@entrypoint(store=store)
def workflow(value: int):
return my_task(value).result()
workflow.invoke(1)3