Example: LangGraph — code review bot
Use case: A code review bot that reads a pull request diff, analyses it for bugs, security issues, and style problems, and posts structured feedback. Built in LangGraph, wrapped with Agentspan — one line change.
What Agentspan adds to LangGraph
LangGraph handles your graph — nodes, edges, conditional branching, typed state. Agentspan adds a production execution layer without changing any of that:
- Crash recovery — graph execution runs on the Agentspan server; a process restart picks up the run without re-running completed steps
- Human-in-the-loop — pause at any tool call for human approval, hold state indefinitely server-side, resume cleanly
- Execution history — every run is logged with full inputs, outputs, and timing, browsable at
http://localhost:6767or via CLI - Re-run from history — replay any past run with the same input from the UI
Your graph definition, nodes, edges, and typed state schema stay exactly as written.
Before: plain LangGraph
This is standard LangGraph code. It runs fine on your laptop but has no durability — if the process dies, the run is lost.
import operator
from typing import TypedDict, Annotated
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
# ── Tools ────────────────────────────────────────────────────────────────────
@tool
def read_file(path: str) -> str:
"""Read a file from the repository."""
return open(path).read()
@tool
def get_pr_diff(pr_number: int, repo: str) -> str:
"""Fetch the unified diff for a GitHub pull request."""
import httpx
resp = httpx.get(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}",
headers={"Accept": "application/vnd.github.v3.diff",
"Authorization": f"Bearer {GITHUB_TOKEN}"},
)
return resp.text
@tool
def post_review_comment(pr_number: int, repo: str, body: str, commit_id: str,
path: str, line: int) -> dict:
"""Post an inline review comment on a specific line of a PR."""
import httpx
resp = httpx.post(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}/comments",
headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
json={"body": body, "commit_id": commit_id, "path": path, "line": line},
)
return resp.json()
tools = [read_file, get_pr_diff, post_review_comment]
tool_node = ToolNode(tools)
# ── Model ─────────────────────────────────────────────────────────────────────
model = ChatAnthropic(model="anthropic/claude-sonnet-4-6").bind_tools(tools)
SYSTEM = """You are an expert code reviewer. When reviewing a pull request:
1. Fetch the diff with get_pr_diff
2. Read any relevant context files with read_file
3. Identify: bugs, security issues, missing error handling, style violations
4. Post inline comments with post_review_comment for each finding
5. End with a summary of findings and an overall verdict (approve / request changes)"""
# ── Graph ─────────────────────────────────────────────────────────────────────
class State(TypedDict):
messages: Annotated[list, operator.add]
def agent_node(state: State):
messages = [SystemMessage(content=SYSTEM)] + state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
def should_continue(state: State):
last = state["messages"][-1]
return "tools" if last.tool_calls else END
workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
app = workflow.compile()
# ── Run (plain LangGraph — no durability) ─────────────────────────────────────
result = app.invoke({
"messages": [HumanMessage(content="Review PR #142 in acme-corp/backend")]
})
print(result["messages"][-1].content)
After: wrapped with Agentspan
Pass the compiled LangGraph app directly to runtime.run(). No imports from agentspan.integrations — Agentspan auto-detects LangGraph apps.
from agentspan.agents import AgentRuntime
# was: result = app.invoke({...})
with AgentRuntime() as runtime:
result = runtime.run(app, {
"messages": [HumanMessage(content="Review PR #142 in acme-corp/backend")]
})
print(result.output["messages"][-1].content)
print(f"Run ID: {result.execution_id}")
That’s it. runtime.run() wraps the entire graph execution as a single managed run on the Agentspan server.
What you gain
Crash recovery — if your process dies mid-review (slow PR, long diff), Agentspan restarts it when a new worker comes up. The graph re-runs from the beginning, but the run is tracked and won’t be lost.
Run history — every PR review is stored with inputs, outputs, and timing. Open http://localhost:6767 to browse execution history in the UI, or query via the API.
Re-run — re-submit any past input from the UI to re-run the review with the latest model or updated instructions.
Async variant
import asyncio
from agentspan.agents import run_async
async def review_pr(pr_number: int, repo: str):
result = await run_async(app, {
"messages": [HumanMessage(content=f"Review PR #{pr_number} in {repo}")]
})
return result.output["messages"][-1].content
asyncio.run(review_pr(142, "acme-corp/backend"))
Fire-and-forget for long reviews
from agentspan.agents import start
# Returns immediately — graph runs in the background
handle = start(app, {
"messages": [HumanMessage(content="Review PR #142 in acme-corp/backend")]
})
print(f"Started: {handle.execution_id}")
print("Checking back later...")
# Get result later
result = handle.stream().get_result()
print(result.output["messages"][-1].content)
Checkpointing and observability
LangGraph and Agentspan serve different purposes — they’re complementary, not alternatives.
LangGraph checkpointing (MemorySaver, PostgresSaver) saves graph state at each node boundary, letting a run resume from its last checkpoint if interrupted. This is a within-run recovery mechanism. When you wrap a LangGraph app with Agentspan, compile without a checkpointer — Agentspan takes over the execution lifecycle and the two mechanisms conflict:
# Works with Agentspan: no checkpointer
app = workflow.compile()
# Does not work with Agentspan: checkpointer + AgentRuntime conflict
# app = workflow.compile(checkpointer=MemorySaver()) # don't do this
Agentspan crash recovery operates at the run level: if your worker process dies, Agentspan restarts the graph run when a new worker connects. Use Agentspan’s crash recovery instead of LangGraph’s node-level checkpointing when your graph is wrapped.
LangSmith traces LLM calls — prompts, completions, token counts, latencies per call. It’s a debugging tool for what the model saw and said.
Agentspan tracks at the run level: execution IDs, full input/output, timing, and completion status across all your agents — LangGraph, OpenAI Agents SDK, Google ADK, or native. The history at http://localhost:6767 is for operations visibility and re-runs, not per-call LLM traces.
If you use LangSmith for LLM debugging today, you can keep it — LangSmith traces will still fire for LLM calls inside your wrapped graph. Agentspan adds run management on top.
Testing (without LangGraph or a server)
Use Agentspan’s standard mock_run with your compiled LangGraph app:
from agentspan.agents.testing import mock_run, MockEvent, expect
result = mock_run(
app,
{"messages": [HumanMessage(content="Review PR #1 in test/repo")]},
events=[
MockEvent.tool_call("get_pr_diff", {"pr_number": 1, "repo": "test/repo"}),
MockEvent.tool_result("get_pr_diff", "- def foo():\n+ def foo(x: int):"),
MockEvent.tool_call("post_review_comment", {"pr_number": 1, "repo": "test/repo", "body": "Consider adding a type hint", "commit_id": "abc123", "path": "main.py", "line": 5}),
MockEvent.tool_result("post_review_comment", {"id": 1, "body": "Consider adding a type hint"}),
MockEvent.done("Review complete. Posted 1 comment."),
]
)
expect(result).completed().used_tool("get_pr_diff").used_tool("post_review_comment")