In the enterprise AI landscape of 2026, multi-agent systems are no longer simple linear chains. They are complex, asynchronous, stateful graphs that interact with external APIs, vector databases, and non-deterministic Large Language Models (LLMs). When you introduce asynchronous I/O, distributed state, and LLM non-determinism, traditional testing paradigms break down.
How do you test an async LangGraph node without actually calling a $20/1M token LLM API? How do you assert that your agent correctly updated the graph state after an async tool call?
This article provides an end-to-end guide to writing and running unit tests for asynchronous Python code using pytest-asyncio. We will explore advanced mocking strategies for external services and apply them to a real-world enterprise use case: a FinTech multi-agent LangGraph RAG system with persistent memory.
Part 1: The Enterprise Use Case
Scenario
"Acme FinTech" has built an AI assistant for financial analysts.
The Workflow
An analyst asks a question (for example, "What is the current price of AAPL, and does our compliance policy allow me to trade it today?").
A Router Agent splits the intent.
A Market Data Agent asynchronously fetches real-time stock prices from an external API (such as Bloomberg or Refinitiv).
A Compliance RAG Agent asynchronously queries an internal vector database for trading restrictions.
A Synthesizer Agent combines the data into a final answer.
The State and Memory
The system uses LangGraph's Checkpointer to maintain conversation history across multiple turns.
The Testing Challenge
We cannot hit the real Bloomberg API or the real vector database in CI/CD. We cannot rely on the LLM to consistently route the query the exact same way every time.
We need deterministic, isolated, and fast asynchronous tests.
Part 2: The System Under Test (Production Code)
Before we test, we must define the async multi-agent system.
Note: In modern pytest-asyncio (standard in 2026), we use asyncio_mode = "auto" in our pyproject.toml so we don't have to decorate every test with @pytest.mark.asyncio.
# app.py
import httpx
from typing import TypedDict, Annotated, Sequence, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph.message import add_messages
from langchain_core.tools import tool
# --- 1. Define the State ---
class GraphState(TypedDict):
# add_messages ensures new messages append correctly to the state
messages: Annotated[Sequence[BaseMessage], add_messages]
market_data: str
compliance_context: str
next_node: str
# --- 2. Define Async Tools (External Services) ---
@tool
async def fetch_stock_price_async(ticker: str) -> str:
"""Fetches real-time stock price asynchronously."""
# In production, this is an async httpx call to an external API
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.marketdata.com/v1/price/{ticker}")
data = response.json()
return f"Ticker {ticker} is trading at ${data['price']}."
@tool
async def search_compliance_db_async(query: str) -> str:
"""Searches the enterprise vector database for compliance rules."""
# In production, this is an async call to Pinecone/Milvus
# Mocking the async vector client call
return "Compliance Rule 4A: Analysts cannot trade a stock within 24 hours of fetching its real-time price if it's an earnings blackout period."
# --- 3. Define the Agent Nodes ---
llm = ChatOpenAI(model="gpt-4o", temperature=0)
async def router_node(state: GraphState):
"""Routes the query based on intent."""
# Simplified routing for demonstration
msg = state["messages"][-1].content.lower()
if "price" in msg and "compliance" in msg:
return {"next_node": "market_agent"}
return {"next_node": "synthesizer"}
async def market_agent_node(state: GraphState):
"""Fetches market data and routes to compliance."""
last_msg = state["messages"][-1].content
# Extract ticker (simplified)
ticker = "AAPL" if "AAPL" in last_msg else "MSFT"
# Call the async tool
price_data = await fetch_stock_price_async.ainvoke({"ticker": ticker})
return {"market_data": price_data, "next_node": "compliance_agent"}
async def compliance_agent_node(state: GraphState):
"""Fetches compliance context via RAG."""
last_msg = state["messages"][-1].content
context = await search_compliance_db_async.ainvoke({"query": last_msg})
return {"compliance_context": context, "next_node": "synthesizer"}
async def synthesizer_node(state: GraphState):
"""Synthesizes the final answer."""
market = state.get("market_data", "No market data.")
compliance = state.get("compliance_context", "No compliance data.")
prompt = [
SystemMessage(content=f"Synthesize an answer. Market: {market}. Compliance: {compliance}"),
state["messages"][-1]
]
# In a real app, we'd use llm.ainvoke. For testability, we'll mock this later.
response = await llm.ainvoke(prompt)
return {"messages": [response], "next_node": "end"}
# --- 4. Build the Graph ---
def route_next(state: GraphState) -> Literal["market_agent", "compliance_agent", "synthesizer", "end"]:
return state["next_node"]
workflow = StateGraph(GraphState)
workflow.add_node("router", router_node)
workflow.add_node("market_agent", market_agent_node)
workflow.add_node("compliance_agent", compliance_agent_node)
workflow.add_node("synthesizer", synthesizer_node)
workflow.set_entry_point("router")
workflow.add_conditional_edges("router", route_next)
workflow.add_edge("market_agent", "compliance_agent")
workflow.add_edge("compliance_agent", "synthesizer")
workflow.add_edge("synthesizer", END)
# Compile with Memory
memory = MemorySaver()
app_graph = workflow.compile(checkpointer=memory)
Part 3: Mocking Strategies for Async External Services
When testing async multi-agent systems, we must isolate the logic from three major sources of non-determinism and latency.
The LLM
LLMs are non-deterministic. If we use a real LLM, the Router might route differently on one run versus another.
Strategy: Mock llm.ainvoke to return pre-defined AIMessage objects.
Async HTTP Clients
We cannot call external APIs during CI.
Strategy: Use unittest.mock.patch together with AsyncMock to intercept httpx.AsyncClient.get.
Async Vector Databases
Strategy: Mock the underlying asynchronous query method of the vector database client.
Part 4: Writing the Tests
The following test suite uses pytest-asyncio.
# test_app.py
import pytest
from unittest.mock import patch, AsyncMock, MagicMock
from langchain_core.messages import HumanMessage, AIMessage
import httpx
# Import our production code
from app import (
app_graph, fetch_stock_price_async, search_compliance_db_async,
market_agent_node, GraphState
)
# ==========================================
# Strategy 1: Unit Testing Async Tools
# ==========================================
@pytest.mark.asyncio
async def test_fetch_stock_price_async_tool():
"""Tests the async HTTP tool by mocking the httpx client."""
# Mock the async response
mock_response = MagicMock()
mock_response.json.return_value = {"price": 150.25}
# We patch the get method of httpx.AsyncClient
with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
mock_get.return_value = mock_response
# Invoke the tool
result = await fetch_stock_price_async.ainvoke({"ticker": "AAPL"})
# Assertions
assert result == "Ticker AAPL is trading at $150.25."
mock_get.assert_awaited_once_with("https://api.marketdata.com/v1/price/AAPL")
# ==========================================
# Strategy 2: Unit Testing a LangGraph Node
# ==========================================
@pytest.mark.asyncio
async def test_market_agent_node_state_update():
"""Tests that the market agent correctly updates the graph state."""
# Mock the async tool inside the node
with patch("app.fetch_stock_price_async.ainvoke", new_callable=AsyncMock) as mock_tool:
mock_tool.return_value = "Ticker AAPL is trading at $150.25."
# Create a mock initial state
initial_state = {
"messages": [HumanMessage(content="What is the price of AAPL?")],
"market_data": "",
"compliance_context": "",
"next_node": ""
}
# Execute the node
result = await market_agent_node(initial_state)
# Assert state transitions
assert result["market_data"] == "Ticker AAPL is trading at $150.25."
assert result["next_node"] == "compliance_agent"
mock_tool.assert_awaited_once()
# ==========================================
# Strategy 3: Integration Testing the Full Graph
# ==========================================
@pytest.mark.asyncio
async def test_full_graph_integration_with_mocked_llm():
"""
Tests the entire multi-agent graph.
We mock the LLM to ensure deterministic routing and synthesis,
and mock the tools to prevent external network calls.
"""
thread_id = "test_thread_123"
config = {"configurable": {"thread_id": thread_id}}
# 1. Mock the external tools
with patch("app.fetch_stock_price_async.ainvoke", new_callable=AsyncMock) as mock_market, \
patch("app.search_compliance_db_async.ainvoke", new_callable=AsyncMock) as mock_compliance:
mock_market.return_value = "AAPL is $150."
mock_compliance.return_value = "No blackout period."
# 2. Mock the LLM to force deterministic behavior
with patch("app.llm.ainvoke", new_callable=AsyncMock) as mock_llm:
mock_llm.return_value = AIMessage(content="AAPL is $150 and you are cleared to trade.")
# 3. Invoke the graph
inputs = {
"messages": [HumanMessage(content="What is AAPL price and can I trade it?")],
"market_data": "",
"compliance_context": "",
"next_node": ""
}
final_state = await app_graph.ainvoke(inputs, config)
# 4. Assertions
assert final_state["messages"][-1].content == "AAPL is $150 and you are cleared to trade."
mock_market.assert_awaited_once()
mock_compliance.assert_awaited_once()
mock_llm.assert_awaited_once()
# ==========================================
# Strategy 4: Testing Memory and State Persistence
# ==========================================
@pytest.mark.asyncio
async def test_graph_memory_persistence():
"""Verifies that the LangGraph Checkpointer correctly maintains state across turns."""
thread_id = "memory_test_thread"
config = {"configurable": {"thread_id": thread_id}}
with patch("app.llm.ainvoke", new_callable=AsyncMock) as mock_llm, \
patch("app.fetch_stock_price_async.ainvoke", new_callable=AsyncMock), \
patch("app.search_compliance_db_async.ainvoke", new_callable=AsyncMock):
# Mock LLM responses for Turn 1 and Turn 2
mock_llm.side_effect = [
AIMessage(content="Turn 1: AAPL is $150."),
AIMessage(content="Turn 2: Yes, you can trade it.")
]
# Turn 1
inputs_turn1 = {
"messages": [HumanMessage(content="What is AAPL price?")],
"market_data": "",
"compliance_context": "",
"next_node": ""
}
await app_graph.ainvoke(inputs_turn1, config)
# Turn 2
inputs_turn2 = {
"messages": [HumanMessage(content="Can I trade it?")],
"market_data": "",
"compliance_context": "",
"next_node": ""
}
final_state = await app_graph.ainvoke(inputs_turn2, config)
# Assert conversation history
assert len(final_state["messages"]) == 4
assert final_state["messages"][0].content == "What is AAPL price?"
assert final_state["messages"][-1].content == "Turn 2: Yes, you can trade it."
Part 5: Configuration and Execution
To run these tests without decorating every asynchronous test function, configure pytest-asyncio in your pyproject.toml.
# pyproject.toml
[tool.pytest.ini_options]
# "auto" mode automatically detects and runs async tests
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
Run the tests using standard pytest.
pytest test_app.py -v
Expected output:
test_app.py::test_fetch_stock_price_async_tool PASSED
test_app.py::test_market_agent_node_state_update PASSED
test_app.py::test_full_graph_integration_with_mocked_llm PASSED
test_app.py::test_graph_memory_persistence PASSED
Part 6: Enterprise Best Practices and Pitfalls
Never Mock the Graph, Mock the Boundaries
Do not mock the LangGraph nodes themselves when performing integration testing.
Instead, mock the boundaries:
The LLM
External tools
External services
This ensures your routing logic, state reducers (add_messages), and checkpointer are actually exercised during testing.
AsyncMock vs. MagicMock
A common mistake is using MagicMock for an asynchronous function.
Doing so results in an error similar to:
TypeError: object MagicMock can't be used in 'await' expression
Always use AsyncMock (or new_callable=AsyncMock when patching) for async def functions.
Event Loop Scoping
By default, pytest-asyncio creates a new event loop for each test.
If your LangGraph checkpointer or database connections depend on a persistent event loop, consider changing the scope to "session" or "module" using @pytest.fixture(scope="session") together with loop_scope="session".
Deterministic LLM Mocking
The side_effect parameter of AsyncMock is ideal for multi-turn agent testing.
It allows you to define a sequence of LLM responses, making conversational tests deterministic without consuming API tokens or introducing LLM variability into CI pipelines.
Use MemorySaver for Tests and PostgresSaver for Production
In this example, MemorySaver is used because it is lightweight and requires no infrastructure.
To validate your production persistence layer, use Testcontainers during CI/CD to start an ephemeral PostgreSQL instance and test against AsyncPostgresSaver.
Conclusion
Testing asynchronous, multi-agent AI systems requires a shift in mindset. Rather than testing simple input-to-output functions, you are validating asynchronous execution, graph state transitions, persistent memory, and interactions with non-deterministic components.
By combining pytest-asyncio for asynchronous execution, AsyncMock for isolating external dependencies, and deterministic LLM mocking for predictable behavior, you can build reliable test suites that give engineering teams confidence when deploying enterprise-scale LangGraph RAG systems into production.
Join the conversation! Your thoughts help the community grow.