LangChain and LangGraph with MonoRouter: Zero-Config Agent Routing

LangChain and LangGraph are the most popular frameworks for building AI agents in Python. Both use the Anthropic SDK under the hood, which means they work with MonoRouter by setting two environment variables — no code changes required.

Quick Start

Set these in your shell before running any LangChain code:

export ANTHROPIC_API_KEY="mr_k_your_monorouter_key"
export ANTHROPIC_BASE_URL="https://api.monorouter.dev/v1"

That's it. Every ChatAnthropic call now routes through MonoRouter.

LangChain: Basic Chat

Here's a minimal example — a summarization chain that works identically whether you're hitting the Anthropic API or MonoRouter:

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    temperature=0,
    max_tokens=1024,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a senior engineer. Summarize the following code change."),
    ("human", "{diff}"),
])

chain = prompt | llm

result = chain.invoke({"diff": open("latest.patch").read()})
print(result.content)

Notice there's no base_url parameter. The SDK picks up ANTHROPIC_BASE_URL from the environment automatically.

LangChain: Structured Output

Pydantic-based structured output works through MonoRouter without any changes. The tool-use protocol passes through transparently:

from pydantic import BaseModel, Field

class CodeReview(BaseModel):
    summary: str = Field(description="One-line summary of the change")
    risk_level: str = Field(description="low, medium, or high")
    issues: list[str] = Field(description="List of potential issues")

structured_llm = llm.with_structured_output(CodeReview)
review = structured_llm.invoke("Review this function: def transfer(amount): db.execute(f'UPDATE accounts SET balance = balance - {amount}')")

print(f"Risk: {review.risk_level}")
for issue in review.issues:
    print(f"  - {issue}")

LangGraph: Building an Agent

LangGraph agents make multiple Claude calls per query — exactly the workload where MonoRouter saves the most money. Here's a practical example: a file-search agent that can read directories and files to answer questions about a codebase.

from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
import os

@tool
def list_directory(path: str) -> str:
    """List files in a directory."""
    try:
        entries = os.listdir(path)
        return "\n".join(entries[:50])
    except OSError as e:
        return f"Error: {e}"

@tool
def read_file(path: str) -> str:
    """Read a file's contents (first 200 lines)."""
    try:
        with open(path) as f:
            lines = f.readlines()[:200]
        return "".join(lines)
    except OSError as e:
        return f"Error: {e}"

llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)
agent = create_react_agent(llm, [list_directory, read_file])

result = agent.invoke({
    "messages": [("human", "What does the auth middleware do in this project?")]
})

print(result["messages"][-1].content)

A single query like this might trigger 5–10 tool calls as the agent navigates the codebase. On the Anthropic API, that's 5–10 billed requests. Through MonoRouter, it's covered by your existing subscription.

Streaming with astream_events

LangGraph's event streaming lets you monitor every step of agent execution in real time. MonoRouter preserves the full SSE stream end-to-end:

import asyncio

async def run_agent():
    llm = ChatAnthropic(model="claude-sonnet-4-20250514", streaming=True)
    agent = create_react_agent(llm, [list_directory, read_file])

    async for event in agent.astream_events(
        {"messages": [("human", "Find all API route handlers")]},
        version="v2",
    ):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                print(chunk.content, end="", flush=True)
        elif kind == "on_tool_start":
            print(f"\n🔧 Calling {event['name']}...")
        elif kind == "on_tool_end":
            print(f"   ✓ Done")

asyncio.run(run_agent())

Token Tracking

LangChain's callback system reports accurate token counts through MonoRouter. Use it to monitor your usage:

from langchain_community.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = chain.invoke({"diff": diff_text})
    print(f"Tokens used: {cb.total_tokens}")
    print(f"  Input: {cb.prompt_tokens}")
    print(f"  Output: {cb.completion_tokens}")

Error Handling

LangChain's built-in retry logic handles MonoRouter's rate limit responses (429) automatically with exponential backoff. No custom error handling needed. If all sessions in your pool are busy, MonoRouter queues the request and LangChain waits for the response — your agent keeps running without interruption.