Deep Research Agent

• Python 3.13
• pip install -U langgraph-cli langchain-anthropic langchain-mcp-adapters tavily-python trustcall
• Node.js + npx (only required for the agent_mcp graph in 3 - Research Agent with MCP)

.env must contain:

ANTHROPIC_API_KEY=sk-ant-...
TAVILY_API_KEY=tvly-...
LANGSMITH_API_KEY="lsv2_pt_..." # Optional
LANGSMITH_TRACING=true # Optional
LANGSMITH_PROJECT="deep-research-agent" # optional

LangSmith keys are optional and live in the LangGraph Studio deployment section below.

• User Clarification - forces the model to emit a typed need_clarification: bool based on initial analysis results. A conditional flow asks user for clarification on their request.
• Human Approval - Human can send approval, or provide feedback to tweak the research brief.
• Public vs internal schema - StateGraph(AgentState, input_schema=AgentInputState) exposes only messages to callers, keeping Studio's input panel clean. Internal state carries research_brief, reviewer_feedback, supervisor_messages, notes, final_report.
• LangSmith evals - notebook runs two LLM-as-judge evals over a synthetic dataset: a success-criteria eval (does the brief capture every requirement the user mentioned?) and a hallucination eval (does the brief invent constraints the user never stated?).

Core pattern - Command routing on structured output:

def clarify_user_request(state: AgentState) -> Command[Literal["write_research_brief", "__end__"]]:
 response = model.with_structured_output(ClarifyUserRequest).invoke(...)
 if response.need_clarification:
 return Command(goto=END, update={"messages": [AIMessage(content=response.question_to_user)]})
 return Command(goto="write_research_brief", update={"messages": [AIMessage(content=response.verification)]})

Core pattern - dynamic interrupt resume contract:

review = interrupt({"type": "brief_review", "research_brief": state["research_brief"], ...})
if review.get("approved"):
 return Command(goto=END)
return Command(goto="write_research_brief", update={"reviewer_feedback": review.get("feedback")})

• Two-tool ReAct surface - tavily_search for retrieval, think_tool for reflection. Reflection is modelled as a tool call so it shows up in the trace next to search results, not buried in a system message.
• Tool-budget cap (MAX_TOOL_ITERATIONS=3) - enforced inside llm_call, not in the conditional edge. At the cap, the unbound model (no tools attached) is invoked with a budget-exhausted nudge so it emits a final synthesis.
• Per-tool observation cap (MAX_OBS_CHARS=30000) - each ToolMessage.content is truncated so accumulated researcher_messages cannot exceed the 200k context window across budgeted iterations.
• compress_research post-loop - a separate model with max_tokens=32000 summarises the entire researcher_messages history. Emits compressed_research (clean handoff) and raw_notes (full dump for downstream auditing).
• Output-schema compile - StateGraph(ResearcherState, output_schema=ResearcherOutputState) filters caller-facing keys to {compressed_research, raw_notes, researcher_messages}, hiding the iteration counter and tool scratchpad.

graph LR
 S[START] --> L[llm_call]
 L -->|tool_calls| T[tool_node]
 T --> L
 L -->|no tool_calls or at cap| C[compress_research]
 C --> E[END]

Core pattern - cap enforced inside llm_call, not the edge:

def llm_call(state: ResearcherState):
 at_cap = state.get("tool_call_iterations", 0) >= MAX_TOOL_ITERATIONS
 chosen_model = model if at_cap else model_with_tools # unbound -> forces synthesis
 return {"researcher_messages": [chosen_model.invoke([SystemMessage(...)] + state["researcher_messages"])]}
def should_continue(state: ResearcherState) -> Literal["tool_node", "compress_research"]:
 return "tool_node" if state["researcher_messages"][-1].tool_calls else "compress_research"

Reuses topology from 2 - Research Agent - START -> llm_call <-> tool_node -> compress_research -> END. The graph wiring is identical; only the tool source and the async semantics change.

New here:

• MultiServerMCPClient over stdio - mcp_config declares command="npx" + args=["-y", "@modelcontextprotocol/server-filesystem", str(current_dir() / "files")]. The filesystem server runs as a managed subprocess scoped to one directory (sandboxed read/write surface).
• Lazy client init - get_mcp_client() and _get_mcp_tools() defer the subprocess spawn until the first node call. Required for LangGraph Platform: the import-time module body must not start subprocesses. Also caches the tool list since the schema is static for the session.
• Async throughout - both llm_call and tool_node are async def. The model is invoked via ainvoke, MCP tools via await tool.ainvoke(...). Forced by MCP's async-stdio transport to the subprocess.
• Mixed sync / async dispatch - think_tool is a sync LangChain tool, MCP tools are async. The dispatcher branches on tool_call["name"] == "think_tool" to pick between tool.invoke(...) and await tool.ainvoke(...).

Core pattern - lazy MCP client + cached tool list:

mcp_config = {"filesystem": {"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", str(current_dir() / "files")], "transport": "stdio"}}
_client = None
_mcp_tools_cache = None
async def _get_mcp_tools():
 global _mcp_tools_cache
 if _mcp_tools_cache is None:
 _mcp_tools_cache = await get_mcp_client().get_tools()
 return _mcp_tools_cache
async def llm_call(state):
 tools = await _get_mcp_tools() + [think_tool]
 return {"researcher_messages": [await model.bind_tools(tools).ainvoke(...)]}

• Tools as decisions - ConductResearch (delegate one sub-research task) and ResearchComplete (declare done) are Pydantic schemas bound as tools to the supervisor model. The supervisor's "decision" is whichever tool it calls.
• Sync + async tools in one turn - the supervisor can emit think_tool (sync), ConductResearch (async sub-agent), and ResearchComplete (terminal sentinel) in the same response. supervisor_tools partitions calls by name, runs the sync batch first, then awaits the async batch.
• Parallel sub-agent fan-out - one researcher_agent.ainvoke(...) per ConductResearch call, all awaited via asyncio.gather(*researcher_invokations). max_concurrent_researchers=2 is enforced through the supervisor prompt (the model is told to emit at most N ConductResearch calls per turn).
• Results as ToolMessages - each sub-agent's compressed_research is wrapped as ToolMessage.content with the matching tool_call_id. get_notes_from_tool_calls(supervisor_messages) later filters tool messages and pulls the strings. This is how findings flow back into supervisor state without a custom reducer.
• Three exit conditions - research_iterations >= max_research_iterations=3, no tool calls on the supervisor's last turn, or any ResearchComplete call. All route to END(of this sub-graph in full-agent setup); otherwise loop back to supervisor.
• raw_notes accumulation - declared as Annotated[list, operator.add] on SupervisorState so per-iteration batches concatenate instead of overwrite.

Core pattern - parallel fan-out + ToolMessage aggregation:

researcher_invokations = [
 researcher_agent.ainvoke({
 "researcher_messages": [HumanMessage(content=tc["args"]["research_topic"])],
 "research_topic": tc["args"]["research_topic"],
 })
 for tc in conduct_research_calls
]
tool_results = await asyncio.gather(*researcher_invokations)
research_tool_messages = [
 ToolMessage(content=r.get("compressed_research", ""), name=tc["name"], tool_call_id=tc["id"])
 for r, tc in zip(tool_results, conduct_research_calls)
]

• Subgraph-as-node composition - "scope_subgraph" & "supervisor_subgraph" agents plug the compiled child graphs in directly. The parent's AgentState is a superset of both child states, so shared keys (messages, research_brief, supervisor_messages, notes) flow across boundaries with no transform layer.
• route_after_scope conditional edge - the scope subgraph terminates two ways: clarification asked (no brief) or HITL approved (brief set). Reading state.get("research_brief") decides whether to run supervisor_subgraph or end the parent run. A clarify-and-wait turn returns control to the caller without firing research.
• Interrupt propagation across boundaries - the scope subgraph's interrupt() surfaces in the parent's checkpoint as a tasks[0].interrupts[0] payload. The caller resumes via the parent's config (Command(resume={"approved": True})); the resume routes back into the subgraph's paused node automatically.
• Async final node - generate_final_report uses research_brief + joined notes into final_report.
• recursion_limit=50 - bumped from the default 25 (set by LangGraph) to fit the complex research loop/jump requirements.

Core pattern - subgraph composition + conditional routing on shared state:

def route_after_scope(state: AgentState) -> Literal["supervisor_subgraph", "__end__"]:
 return "supervisor_subgraph" if state.get("research_brief") else END
builder = StateGraph(AgentState, input_schema=AgentInputState)
builder.add_node("scope_subgraph", scope_research_agent) # compiled subgraph
builder.add_node("supervisor_subgraph", supervisor_agent) # compiled subgraph
builder.add_node("final_report_generation", generate_final_report)
builder.add_edge(START, "scope_subgraph")
builder.add_conditional_edges("scope_subgraph", route_after_scope, ["supervisor_subgraph", END])
builder.add_edge("supervisor_subgraph", "final_report_generation")
builder.add_edge("final_report_generation", END)

Add to .env:

LANGSMITH_API_KEY=lsv2_pt_...
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=deep-research-agent

After langgraph dev boots, connect Studio at:

https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

Note: the Lite control plane requires LANGSMITH_API_KEY even when LANGSMITH_TRACING=false (see project CLAUDE.md).

cd 100_Agents/1_Deep_Research_Agent
langgraph dev

The CLI watches src/ and hot-reloads on save. Stop with Ctrl+C.

section_scope_research_agent (input schema: AgentInputState, key messages):

Turn 1 - vague request, clarify branch:

{ "messages": [{ "role": "user", "content": "Agentic AI best practices" }] }

clarify_user_request decides need_clarification=True, appends an AIMessage with a clarification question, and routes to END. The graph terminates without writing a brief.

Turn 2 - reply on the same thread:

In Studio, keep the same thread open and submit the user's answer as the next turn. The add_messages reducer appends it to the existing message history; clarify_user_request re-runs against the full conversation:

{ "messages": [{ "role": "user", "content": "Production deployments. Focus on observability, evals, and guardrails for tool-using agents in regulated industries." }] }

This time need_clarification=False, control flows to write_research_brief -> human_review_brief, and the graph pauses on the interrupt.

Turn 3 - resume the interrupt:

{ "approved": true }

Or reject with feedback to loop back into write_research_brief in revise mode:

{ "approved": false, "feedback": "Tighten scope to evaluation harnesses only; drop guardrails." }

section_researcher_agent (input schema: ResearcherState, keys researcher_messages + research_topic):

{
 "researcher_messages": [
 { "role": "user", "content": "Research production best practices for agentic AI: observability, evals, guardrails." }
 ],
 "research_topic": "Production best practices for agentic AI: observability, evals, guardrails"
}

Expect Tavily search calls and a compressed_research field on the output.

section_agent_mcp (same schema as section_researcher_agent; requires src/files/ populated with at least one document):

{
 "researcher_messages": [
 { "role": "user", "content": "Summarise the documents under src/files/." }
 ],
 "research_topic": "Local research notes summary"
}

section_supervisor_agent (input schema: SupervisorState, key supervisor_messages):

{
 "supervisor_messages": [
 { "role": "user", "content": "I'm researching system design and architecture for production agentic AI: state, orchestration, evals." }
 ]
}

Watch for parallel researcher sub-agents firing via asyncio.gather.

full_deep_researcher_agent (input schema: AgentInputState, key messages; full pipeline with HITL gate):

This graph embeds section_scope_research_agent as a subgraph at the start, so the same multi-turn clarification flow applies before any research runs.

Turn 1 - vague kickoff, clarify branch terminates the run:

{ "messages": [{ "role": "user", "content": "I want to learn about AI agents." }] }

The scope subgraph asks a clarification question and the parent graph's route_after_scope sees no research_brief, so it routes to END.

Turn 2 - reply on the same thread:

{ "messages": [{ "role": "user", "content": "Specifically, security best practices: prompt-injection defenses, tool sandboxing, and least-privilege credential handling for production agents." }] }

clarify_user_request re-runs, accepts the brief, and the graph pauses at human_review_brief inside the subgraph.

Turn 3 - resume the interrupt:

{ "approved": true }

or reject with feedback (loops back to write_research_brief in revise mode):

{ "approved": false, "feedback": "Focus only on prompt-injection defenses; drop the credential section." }

After approval, control returns to the parent graph: supervisor_subgraph fans out research, then final_report_generation writes the report into final_report.

Skip-clarification shortcut: if Turn 1 is detailed enough, clarify_user_request returns need_clarification=False immediately, the run goes straight to human_review_brief, and Turn 2 is the resume payload.

{ "messages": [{ "role": "user", "content": "Research security best practices for production AI agents: prompt-injection defenses, tool-use sandboxing, and credential handling." }] }