Integrating ClaudeCLIChatModel with LangGraph

ClaudeCLIChatModel and LangGraph Integration

I wanted to connect Claude Control sessions to the LangChain/LangGraph ecosystem. LangGraph's StateGraph lets you define complex agent workflows, and pairing it with a Claude CLI session as the LLM backend makes for a powerful combination.

Implementing ClaudeCLIChatModel

I subclassed LangChain's BaseChatModel to wrap a Claude CLI session as an LLM.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from typing import List, Optional, Any

class ClaudeCLIChatModel(BaseChatModel):
    session_manager: Any
    session_id: str
    model_name: str = "claude-cli"

    class Config:
        arbitrary_types_allowed = True

    @property
    def _llm_type(self) -> str:
        return "claude-cli"

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        **kwargs,
    ) -> ChatResult:
        raise NotImplementedError("Use async version")

    async def _agenerate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        **kwargs,
    ) -> ChatResult:
        # Convert messages into a single prompt
        prompt = self._format_messages(messages)
        response = await self.session_manager.send_prompt(
            self.session_id, prompt
        )
        message = AIMessage(content=response)
        generation = ChatGeneration(message=message)
        return ChatResult(generations=[generation])

    def _format_messages(self, messages: List[BaseMessage]) -> str:
        parts = []
        for msg in messages:
            if isinstance(msg, HumanMessage):
                parts.append(f"사용자: {msg.content}")
            elif isinstance(msg, AIMessage):
                parts.append(f"어시스턴트: {msg.content}")
            else:
                parts.append(msg.content)
        return "\n".join(parts)

Using LangGraph StateGraph

I used LangGraph's state graph to define a code review workflow.

from langgraph.graph import StateGraph, END
from typing import TypedDict

class ReviewState(TypedDict):
    code: str
    review_comments: list
    is_approved: bool
    revision_count: int

async def analyze_code(state: ReviewState) -> ReviewState:
    llm = ClaudeCLIChatModel(
        session_manager=manager,
        session_id="review-session",
    )
    response = await llm.ainvoke([
        HumanMessage(content=f"다음 코드를 분석하세요:\n{state['code']}")
    ])
    state["review_comments"].append(response.content)
    return state

async def check_approval(state: ReviewState) -> str:
    if state["is_approved"] or state["revision_count"] > 3:
        return "end"
    return "revise"

graph = StateGraph(ReviewState)
graph.add_node("analyze", analyze_code)
graph.add_node("revise", revise_code)
graph.add_conditional_edges("analyze", check_approval, {
    "end": END,
    "revise": "revise",
})
graph.add_edge("revise", "analyze")
graph.set_entry_point("analyze")

workflow = graph.compile()

Results in Practice

With the Claude CLI session running as a LangGraph node, I was able to fully automate the review → revise → re-review loop. Compared to direct API calls, the key advantages are that Claude CLI has direct filesystem access and tool use feels natural.

Troubleshooting

The biggest pain point was reconciling LangChain's sync and async interfaces. You need to implement both _generate and _agenerate, but Claude CLI sessions are inherently async — calling asyncio.run() inside the sync version causes event loop conflicts. I ended up settling on async-only usage.