How do you build stateful tools that read/write to a database across multiple agent turns?

Question

Accepted Answer

To build stateful tools that read from or write to a database across multiple agent turns, you need two components: a persistence layer for the agent's conversation state (checkpointer) and a shared resource (database connection) that your tools can access. The checkpointer saves the entire agent state between turns, including messages and any custom fields, allowing the agent to resume exactly where it left off[citation:1]. The tools can then use a shared database connection passed via configuration or dependency injection to store user preferences, session data, or any other information that must persist across turns. ```python
from langgraph.graph import StateGraph, MessagesState
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from typing import Dict, Any

# Extend state to include database results
class AppState(MessagesState):
    results: Dict[str, Any]  # Store query results across turns
    user_preferences: Dict[str, Any]  # Persist user settings

# Tool that reads from database using shared connection
@tool
def get_user_preferences(user_id: str, config: RunnableConfig) -> str:
    """Retrieve user preferences from the database."""
    db = config.get("configurable", {}).get("db_connection")
    result = db.query("SELECT * FROM preferences WHERE user_id = ?", [user_id])
    return str(result)

# Tool that writes to database
@tool
def save_note(note_content: str, config: RunnableConfig) -> str:
    """Save a note to the database for this conversation."""
    thread_id = config.get("configurable", {}).get("thread_id")
    db = config.get("configurable", {}).get("db_connection")
    db.execute("INSERT INTO notes (thread_id, content) VALUES (?, ?)", [thread_id, note_content])
    return f"Note saved: {note_content}"

# Set up checkpointer for persistence across turns
checkpointer = SqliteSaver.from_conn_string("sqlite:///conversations.db")

# Create agent with checkpointer
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[get_user_preferences, save_note],
    checkpointer=checkpointer
)

# Use thread_id to maintain separate conversation state
config = {"configurable": {"thread_id": "user-123", "db_connection": db}}

# First turn: agent can save data
result = agent.invoke({"messages": [("human", "Remember that I like sci-fi movies")]}, config)

# Second turn: agent recalls saved data using same thread_id
result = agent.invoke({"messages": [("human", "What movies do I like?")]}, config)
```

How do you build stateful tools that read/write to a database across multiple agent turns?

Build stateful tools that persist data across agent turns by using a checkpointer to save the entire agent state and a shared database connection accessible to your tools, which can read and write based on conversation history stored in the state.

How do you build stateful tools that read/write to a database across multiple agent turns?

Build stateful tools that persist data across agent turns by using a checkpointer to save the entire agent state and a shared database connection accessible to your tools, which can read and write based on conversation history stored in the state.

Scenario Questions

Follow-up Questions