Implementation:Langchain ai Langgraph Config Accessors

Knowledge Sources	Langchain_ai_Langgraph
Domains	Configuration, Runtime
Last Updated	2026-02-11 16:00 GMT

Overview

Provides runtime accessor functions (`get_config`, `get_store`, `get_stream_writer`) that allow graph nodes and tasks to retrieve the current execution configuration, persistent store, and stream writer from context variables.

Description

The Config Accessors module exposes three functions that enable LangGraph graph nodes and functional API tasks to access runtime context without requiring these objects to be passed explicitly through function signatures. All three functions rely on LangChain's `var_child_runnable_config` context variable to retrieve the current `RunnableConfig`, which contains the runtime configuration injected by the Pregel execution engine.

The `get_config()` function retrieves the full `RunnableConfig` dictionary from the current context. It includes a Python version check that raises a `RuntimeError` if called in an async context on Python < 3.11, since `contextvars` propagation to child tasks only became reliable in Python 3.11. If called outside of a runnable context (i.e., when no config is available in the context variable), it raises a `RuntimeError`.

The `get_store()` function extracts the `BaseStore` instance from the runtime configuration. It navigates through the config's internal `CONF` and `CONFIG_KEY_RUNTIME` keys to access the store object. This requires that the graph was compiled with a store (e.g., `graph.compile(store=store)`). The `get_stream_writer()` function similarly extracts the `StreamWriter` callable from the runtime configuration, enabling nodes to emit custom streaming data via the `"custom"` stream mode.

Usage

Use these accessor functions inside any `StateGraph` node function or functional API `@task` decorated function to access runtime services. `get_store()` is used to read and write persistent data from within nodes, while `get_stream_writer()` enables custom streaming output. `get_config()` provides access to the full configuration for advanced use cases like reading configurable parameters or metadata.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/langgraph/langgraph/config.py

Signature

def get_config() -> RunnableConfig: ...

def get_store() -> BaseStore: ...

def get_stream_writer() -> StreamWriter: ...

Import

from langgraph.config import get_config, get_store, get_stream_writer

I/O Contract

get_config

Input	Output	Error Conditions
(none)	`RunnableConfig`	`RuntimeError` if called outside a runnable context
(none)	`RunnableConfig`	`RuntimeError` if called in async context on Python < 3.11

get_store

Input	Output	Error Conditions
(none)	`BaseStore`	`RuntimeError` if called outside a runnable context (via `get_config`)
(none)	`BaseStore`	`KeyError` / `AttributeError` if graph was not compiled with a store

get_stream_writer

Input	Output	Error Conditions
(none)	`StreamWriter`	`RuntimeError` if called outside a runnable context (via `get_config`)

Prerequisites

Function	Requirement
`get_config()`	Must be called inside a graph node or `@task` function during execution
`get_store()`	Graph must be compiled with `store=` parameter
`get_stream_writer()`	Graph must be invoked via `.stream()` for the writer to produce output
All (async)	Python >= 3.11 required for async context support

Usage Examples

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.store.memory import InMemoryStore
from langgraph.config import get_config, get_store, get_stream_writer


class State(TypedDict):
    foo: int


# Example 1: Using get_store() to access persistent storage
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})


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)
)

result = graph.invoke({"foo": 1})
print(result)  # {"foo": 3}


# Example 2: Using get_stream_writer() for custom streaming
def streaming_node(state: State):
    writer = get_stream_writer()
    writer({"progress": "processing..."})
    writer({"progress": "done!"})
    return {"foo": state["foo"] + 1}


stream_graph = (
    StateGraph(State)
    .add_node(streaming_node)
    .add_edge(START, "streaming_node")
    .compile(store=store)
)

for chunk in stream_graph.stream({"foo": 1}, stream_mode="custom"):
    print(chunk)
# {"progress": "processing..."}
# {"progress": "done!"}


# Example 3: Using get_config() to access runtime configuration
def config_aware_node(state: State):
    config = get_config()
    thread_id = config.get("configurable", {}).get("thread_id", "unknown")
    print(f"Running in thread: {thread_id}")
    return {"foo": state["foo"] + 1}

Related Pages

• Principle:Langchain_ai_Langgraph_Runtime_Config_Access