0
# State and Store Injection
1

2
Annotations for injecting graph state and persistent storage into tool arguments, enabling context-aware tools without exposing internal state management to the language model.
3

4
## Capabilities
5

6
### InjectedState Annotation
7

8
Annotation for injecting graph state into tool arguments. Enables tools to access graph state without exposing state management details to the language model.
9

10
```python { .api }
11
class InjectedState(InjectedToolArg):
12
    def __init__(self, field: Optional[str] = None) -> None
13
```
14

15
**Parameters:**
16
- `field`: Optional key to extract from the state dictionary. If None, the entire state is injected.
17

18
**Usage Examples:**
19

20
```python
21
from typing import List
22
from typing_extensions import Annotated, TypedDict
23
from langchain_core.messages import BaseMessage
24
from langchain_core.tools import tool
25
from langgraph.prebuilt import InjectedState, ToolNode
26

27
class AgentState(TypedDict):
28
    messages: List[BaseMessage]
29
    user_id: str
30
    session_data: dict
31

32
# Inject entire state
33
@tool
34
def state_aware_tool(
35
    query: str,
36
    state: Annotated[dict, InjectedState]
37
) -> str:
38
    """Tool that accesses full graph state."""
39
    user_id = state.get("user_id", "unknown")
40
    message_count = len(state.get("messages", []))
41

42
    if message_count > 5:
43
        return f"Continuing conversation for user {user_id}: {query}"
44
    else:
45
        return f"Starting fresh conversation for user {user_id}: {query}"
46

47
# Inject specific state field
48
@tool
49
def user_specific_tool(
50
    action: str,
51
    user_id: Annotated[str, InjectedState("user_id")]
52
) -> str:
53
    """Tool that accesses specific state field."""
54
    return f"Performing {action} for user {user_id}"
55

56
# Create tool node
57
tool_node = ToolNode([state_aware_tool, user_specific_tool])
58

59
# Example state
60
state = {
61
    "messages": [BaseMessage(content="Hello"), BaseMessage(content="How are you?")],
62
    "user_id": "user_123",
63
    "session_data": {"theme": "dark", "language": "en"}
64
}
65

66
# Tool calls with state injection
67
tool_calls = [
68
    {"name": "state_aware_tool", "args": {"query": "weather"}, "id": "1", "type": "tool_call"},
69
    {"name": "user_specific_tool", "args": {"action": "save_preference"}, "id": "2", "type": "tool_call"}
70
]
71

72
# State is automatically injected
73
result = tool_node.invoke({"messages": [...], **state})
74
```
75

76
### InjectedStore Annotation
77

78
Annotation for injecting persistent store into tool arguments. Enables tools to access LangGraph's persistent storage system for cross-session data persistence.
79

80
```python { .api }
81
class InjectedStore(InjectedToolArg):
82
    pass
83
```
84

85
**Usage Examples:**
86

87
```python
88
from typing_extensions import Annotated
89
from langchain_core.tools import tool
90
from langgraph.prebuilt import InjectedStore, ToolNode
91
from langgraph.store.base import BaseStore
92

93
@tool
94
def save_user_preference(
95
    key: str,
96
    value: str,
97
    user_id: str,
98
    store: Annotated[BaseStore, InjectedStore()]
99
) -> str:
100
    """Save user preference to persistent storage."""
101
    namespace = ("user_preferences", user_id)
102
    store.put(namespace, key, value)
103
    return f"Saved {key} = {value} for user {user_id}"
104

105
@tool
106
def get_user_preference(
107
    key: str,
108
    user_id: str,
109
    store: Annotated[BaseStore, InjectedStore()]
110
) -> str:
111
    """Retrieve user preference from persistent storage."""
112
    namespace = ("user_preferences", user_id)
113
    result = store.get(namespace, key)
114
    return result.value if result else f"No preference found for {key}"
115

116
@tool
117
def list_user_data(
118
    user_id: str,
119
    store: Annotated[BaseStore, InjectedStore()]
120
) -> str:
121
    """List all stored data for a user."""
122
    namespace = ("user_preferences", user_id)
123
    items = store.search(namespace)
124
    if items:
125
        return f"Stored preferences: {', '.join(item.key for item in items)}"
126
    return "No stored preferences found"
127

128
# Usage with graph compilation
129
from langgraph.graph import StateGraph
130
from langgraph.store.memory import InMemoryStore
131

132
store = InMemoryStore()
133
tool_node = ToolNode([save_user_preference, get_user_preference, list_user_data])
134

135
graph = StateGraph(AgentState)
136
graph.add_node("tools", tool_node)
137
compiled_graph = graph.compile(store=store)  # Store is injected automatically
138
```
139

140
## Advanced State Injection Patterns
141

142
### Conditional State Access
143

144
```python
145
@tool
146
def adaptive_tool(
147
    query: str,
148
    state: Annotated[dict, InjectedState]
149
) -> str:
150
    """Tool that adapts behavior based on state."""
151
    messages = state.get("messages", [])
152
    user_tier = state.get("user_tier", "basic")
153

154
    # Adapt behavior based on conversation length
155
    if len(messages) < 3:
156
        response_style = "detailed"
157
    elif len(messages) > 10:
158
        response_style = "concise"
159
    else:
160
        response_style = "balanced"
161

162
    # Adapt features based on user tier
163
    advanced_features = user_tier in ["premium", "enterprise"]
164

165
    return f"Processing '{query}' with {response_style} style, advanced features: {advanced_features}"
166
```
167

168
### Multi-Field State Injection
169

170
```python
171
@tool
172
def context_rich_tool(
173
    task: str,
174
    user_id: Annotated[str, InjectedState("user_id")],
175
    session_data: Annotated[dict, InjectedState("session_data")],
176
    messages: Annotated[list, InjectedState("messages")]
177
) -> str:
178
    """Tool with multiple injected state fields."""
179
    user_language = session_data.get("language", "en")
180
    conversation_length = len(messages)
181

182
    return f"Task: {task} | User: {user_id} | Language: {user_language} | Messages: {conversation_length}"
183
```
184

185
## Persistent Storage Patterns
186

187
### Namespaced Data Organization
188

189
```python
190
@tool
191
def save_conversation_summary(
192
    summary: str,
193
    conversation_id: str,
194
    store: Annotated[BaseStore, InjectedStore()]
195
) -> str:
196
    """Save conversation summary with organized namespacing."""
197
    namespace = ("conversations", "summaries")
198
    store.put(namespace, conversation_id, {
199
        "summary": summary,
200
        "created_at": datetime.now().isoformat(),
201
        "version": "1.0"
202
    })
203
    return f"Saved summary for conversation {conversation_id}"
204

205
@tool
206
def get_user_history(
207
    user_id: str,
208
    store: Annotated[BaseStore, InjectedStore()]
209
) -> str:
210
    """Retrieve user's conversation history."""
211
    namespace = ("users", user_id, "history")
212
    items = store.search(namespace)
213
    return f"Found {len(items)} historical items for user {user_id}"
214
```
215

216
### Cross-Session Data Sharing
217

218
```python
219
@tool
220
def update_global_stats(
221
    action: str,
222
    store: Annotated[BaseStore, InjectedStore()]
223
) -> str:
224
    """Update global application statistics."""
225
    namespace = ("global", "stats")
226
    current_stats = store.get(namespace, "counters")
227

228
    if current_stats:
229
        counters = current_stats.value
230
    else:
231
        counters = {}
232

233
    counters[action] = counters.get(action, 0) + 1
234
    store.put(namespace, "counters", counters)
235

236
    return f"Updated {action} count to {counters[action]}"
237
```
238

239
## Integration with ToolNode
240

241
ToolNode automatically handles injection during tool execution:
242

243
```python
244
from langgraph.prebuilt import ToolNode
245

246
# Tools with injected dependencies
247
tools_with_injection = [
248
    save_user_preference,
249
    get_user_preference,
250
    state_aware_tool,
251
    context_rich_tool
252
]
253

254
tool_node = ToolNode(tools_with_injection)
255

256
# Injection happens automatically during invoke()
257
state = {
258
    "messages": [...],
259
    "user_id": "user_123",
260
    "session_data": {"language": "en", "theme": "dark"}
261
}
262

263
# Both state and store are injected as needed
264
result = tool_node.invoke(state, store=my_store)
265
```
266

267
## Manual Injection for Advanced Use Cases
268

269
For custom routing or Send API usage:
270

271
```python
272
# Manually inject arguments for custom routing
273
tool_call = {
274
    "name": "save_user_preference",
275
    "args": {"key": "theme", "value": "dark", "user_id": "user_123"},
276
    "id": "tool_1",
277
    "type": "tool_call"
278
}
279

280
# Inject state and store
281
injected_call = tool_node.inject_tool_args(tool_call, state, store)
282

283
# Use with Send API
284
from langgraph.types import Send
285

286
def custom_router(state):
287
    tool_calls = state["messages"][-1].tool_calls
288
    injected_calls = [
289
        tool_node.inject_tool_args(call, state, store)
290
        for call in tool_calls
291
    ]
292
    return [Send("tools", call) for call in injected_calls]
293
```
294

295
## Error Handling
296

297
### Missing Store Error
298

299
```python
300
# Error occurs if tool requires store injection but none provided
301
@tool
302
def requires_store_tool(
303
    data: str,
304
    store: Annotated[BaseStore, InjectedStore()]
305
) -> str:
306
    return store.get(("data",), "key").value
307

308
tool_node = ToolNode([requires_store_tool])
309

310
# This will raise ValueError about missing store
311
try:
312
    result = tool_node.invoke(state)  # No store provided
313
except ValueError as e:
314
    print(f"Error: {e}")  # "Cannot inject store into tools with InjectedStore annotations"
315
```
316

317
### State Field Missing
318

319
```python
320
@tool
321
def requires_field_tool(
322
    query: str,
323
    user_id: Annotated[str, InjectedState("user_id")]
324
) -> str:
325
    return f"Query for user {user_id}: {query}"
326

327
# If state doesn't contain "user_id" field, KeyError occurs
328
state_without_user_id = {"messages": []}
329

330
try:
331
    result = tool_node.invoke(state_without_user_id)
332
except KeyError as e:
333
    print(f"Missing required state field: {e}")
334
```
335

336
## Best Practices
337

338
### State Design
339

340
```python
341
# Good: Clear, typed state schema
342
class ApplicationState(TypedDict):
343
    messages: List[BaseMessage]
344
    user_id: str
345
    session_metadata: dict
346
    preferences: dict
347

348
# Avoid: Overly nested or unclear state structure
349
```
350

351
### Store Organization
352

353
```python
354
# Good: Consistent namespacing strategy
355
namespace = ("domain", "entity_type", "entity_id")
356
store.put(("users", "preferences", user_id), "theme", "dark")
357
store.put(("conversations", "summaries", conv_id), "summary", text)
358

359
# Good: Version your stored data
360
data = {"value": content, "version": 1, "created_at": timestamp}
361
```
362

363
### Tool Design
364

365
```python
366
# Good: Clear parameter separation
367
@tool
368
def well_designed_tool(
369
    user_input: str,                                    # From model
370
    user_id: Annotated[str, InjectedState("user_id")], # From state
371
    store: Annotated[BaseStore, InjectedStore()]        # From system
372
) -> str:
373
    """Clear separation of concerns."""
374
    pass
375

376
# Avoid: Mixing injected and model parameters confusingly
377
```