tessl/pypi-langgraph-prebuilt

Library with high-level APIs for creating and executing LangGraph agents and tools.

Overview

Eval results

Files

State and Store Injection

Name: tessl/pypi-langgraph-prebuilt
Author: tessl

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

Capabilities

InjectedState Annotation

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

class InjectedState(InjectedToolArg):
    def __init__(self, field: Optional[str] = None) -> None

Parameters:

field: Optional key to extract from the state dictionary. If None, the entire state is injected.

Usage Examples:

from typing import List
from typing_extensions import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState, ToolNode

class AgentState(TypedDict):
    messages: List[BaseMessage]
    user_id: str
    session_data: dict

# Inject entire state
@tool
def state_aware_tool(
    query: str,
    state: Annotated[dict, InjectedState]
) -> str:
    """Tool that accesses full graph state."""
    user_id = state.get("user_id", "unknown")
    message_count = len(state.get("messages", []))

    if message_count > 5:
        return f"Continuing conversation for user {user_id}: {query}"
    else:
        return f"Starting fresh conversation for user {user_id}: {query}"

# Inject specific state field
@tool
def user_specific_tool(
    action: str,
    user_id: Annotated[str, InjectedState("user_id")]
) -> str:
    """Tool that accesses specific state field."""
    return f"Performing {action} for user {user_id}"

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

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

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

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

InjectedStore Annotation

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

class InjectedStore(InjectedToolArg):
    pass

Usage Examples:

from typing_extensions import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedStore, ToolNode
from langgraph.store.base import BaseStore

@tool
def save_user_preference(
    key: str,
    value: str,
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """Save user preference to persistent storage."""
    namespace = ("user_preferences", user_id)
    store.put(namespace, key, value)
    return f"Saved {key} = {value} for user {user_id}"

@tool
def get_user_preference(
    key: str,
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """Retrieve user preference from persistent storage."""
    namespace = ("user_preferences", user_id)
    result = store.get(namespace, key)
    return result.value if result else f"No preference found for {key}"

@tool
def list_user_data(
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """List all stored data for a user."""
    namespace = ("user_preferences", user_id)
    items = store.search(namespace)
    if items:
        return f"Stored preferences: {', '.join(item.key for item in items)}"
    return "No stored preferences found"

# Usage with graph compilation
from langgraph.graph import StateGraph
from langgraph.store.memory import InMemoryStore

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

graph = StateGraph(AgentState)
graph.add_node("tools", tool_node)
compiled_graph = graph.compile(store=store)  # Store is injected automatically

Advanced State Injection Patterns

Conditional State Access

@tool
def adaptive_tool(
    query: str,
    state: Annotated[dict, InjectedState]
) -> str:
    """Tool that adapts behavior based on state."""
    messages = state.get("messages", [])
    user_tier = state.get("user_tier", "basic")

    # Adapt behavior based on conversation length
    if len(messages) < 3:
        response_style = "detailed"
    elif len(messages) > 10:
        response_style = "concise"
    else:
        response_style = "balanced"

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

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

Multi-Field State Injection

@tool
def context_rich_tool(
    task: str,
    user_id: Annotated[str, InjectedState("user_id")],
    session_data: Annotated[dict, InjectedState("session_data")],
    messages: Annotated[list, InjectedState("messages")]
) -> str:
    """Tool with multiple injected state fields."""
    user_language = session_data.get("language", "en")
    conversation_length = len(messages)

    return f"Task: {task} | User: {user_id} | Language: {user_language} | Messages: {conversation_length}"

Persistent Storage Patterns

Namespaced Data Organization

@tool
def save_conversation_summary(
    summary: str,
    conversation_id: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """Save conversation summary with organized namespacing."""
    namespace = ("conversations", "summaries")
    store.put(namespace, conversation_id, {
        "summary": summary,
        "created_at": datetime.now().isoformat(),
        "version": "1.0"
    })
    return f"Saved summary for conversation {conversation_id}"

@tool
def get_user_history(
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """Retrieve user's conversation history."""
    namespace = ("users", user_id, "history")
    items = store.search(namespace)
    return f"Found {len(items)} historical items for user {user_id}"

Cross-Session Data Sharing

@tool
def update_global_stats(
    action: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """Update global application statistics."""
    namespace = ("global", "stats")
    current_stats = store.get(namespace, "counters")

    if current_stats:
        counters = current_stats.value
    else:
        counters = {}

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

    return f"Updated {action} count to {counters[action]}"

Integration with ToolNode

ToolNode automatically handles injection during tool execution:

from langgraph.prebuilt import ToolNode

# Tools with injected dependencies
tools_with_injection = [
    save_user_preference,
    get_user_preference,
    state_aware_tool,
    context_rich_tool
]

tool_node = ToolNode(tools_with_injection)

# Injection happens automatically during invoke()
state = {
    "messages": [...],
    "user_id": "user_123",
    "session_data": {"language": "en", "theme": "dark"}
}

# Both state and store are injected as needed
result = tool_node.invoke(state, store=my_store)

Manual Injection for Advanced Use Cases

For custom routing or Send API usage:

# Manually inject arguments for custom routing
tool_call = {
    "name": "save_user_preference",
    "args": {"key": "theme", "value": "dark", "user_id": "user_123"},
    "id": "tool_1",
    "type": "tool_call"
}

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

# Use with Send API
from langgraph.types import Send

def custom_router(state):
    tool_calls = state["messages"][-1].tool_calls
    injected_calls = [
        tool_node.inject_tool_args(call, state, store)
        for call in tool_calls
    ]
    return [Send("tools", call) for call in injected_calls]

Error Handling

Missing Store Error

# Error occurs if tool requires store injection but none provided
@tool
def requires_store_tool(
    data: str,
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    return store.get(("data",), "key").value

tool_node = ToolNode([requires_store_tool])

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

State Field Missing

@tool
def requires_field_tool(
    query: str,
    user_id: Annotated[str, InjectedState("user_id")]
) -> str:
    return f"Query for user {user_id}: {query}"

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

try:
    result = tool_node.invoke(state_without_user_id)
except KeyError as e:
    print(f"Missing required state field: {e}")

Best Practices

State Design

# Good: Clear, typed state schema
class ApplicationState(TypedDict):
    messages: List[BaseMessage]
    user_id: str
    session_metadata: dict
    preferences: dict

# Avoid: Overly nested or unclear state structure

Store Organization

# Good: Consistent namespacing strategy
namespace = ("domain", "entity_type", "entity_id")
store.put(("users", "preferences", user_id), "theme", "dark")
store.put(("conversations", "summaries", conv_id), "summary", text)

# Good: Version your stored data
data = {"value": content, "version": 1, "created_at": timestamp}

Tool Design

# Good: Clear parameter separation
@tool
def well_designed_tool(
    user_input: str,                                    # From model
    user_id: Annotated[str, InjectedState("user_id")], # From state
    store: Annotated[BaseStore, InjectedStore()]        # From system
) -> str:
    """Clear separation of concerns."""
    pass

# Avoid: Mixing injected and model parameters confusingly

Install with Tessl CLI

npx tessl i tessl/pypi-langgraph-prebuilt

docs

agent-creation.md

human-in-the-loop.md

index.md

state-store-injection.md

tool-execution.md

tool-validation.md

tile.json