How to Add a Custom Tool to a LangGraph Agent Workflow: Complete Implementation Guide

Adding a custom tool to a LangGraph agent workflow requires defining an async function with the @tool decorator, registering it in the ToolNode during graph construction in AgenticRAGService._build_graph, and emitting a matching tool_calls entry from any agent node that triggers the capability.

The production-grade RAG implementation in the jamwithai/production-agentic-rag-course repository demonstrates a scalable pattern for extending LangGraph agents with external capabilities. Adding a custom tool to LangGraph agent workflows follows a consistent three-step architecture that maintains type safety and clear separation of concerns.

Step 1: Define the Tool with @tool Decorator

Custom tool definitions reside in src/services/agents/tools.py, using a factory pattern that accepts service dependencies and returns an async function decorated with @tool from langchain_core.tools.

# src/services/agents/tools.py

# Lines 12-28 demonstrate the factory pattern implementation

def create_retriever_tool(
    opensearch_client: OpenSearchClient,
    embeddings_client: JinaEmbeddingsClient,
    top_k: int = 3,
    use_hybrid: bool = True,
):
    @tool
    async def retrieve_papers(query: str) -> list[Document]:
        """Search and return relevant arXiv research papers."""
        # Embedding, OpenSearch call, and Document conversion logic

        return documents
    return retrieve_papers

Critical implementation requirements:

• Async signature: LangGraph executes nodes asynchronously, so tools must use async def.
• @tool decorator: Registers the function with LangChain's tool registry, enabling the graph to discover and invoke it by name.
• Dependency injection: The factory receives concrete service instances (OpenSearch, embeddings) rather than importing them directly, keeping the tool pure and testable.
• Return types: Functions should return serializable objects like list[Document] that LangGraph can store in state.

Step 2: Register the Tool in the Graph Workflow

The workflow construction happens inside src/services/agents/agentic_rag.py within the AgenticRAGService._build_graph method. Here you instantiate the tool and add it to the ToolNode.

# src/services/agents/agentic_rag.py

# Lines 88-96 instantiate the tool

retriever_tool = create_retriever_tool(
    opensearch_client=self.opensearch,
    embeddings_client=self.embeddings,
    top_k=self.graph_config.top_k,
    use_hybrid=self.graph_config.use_hybrid,
)
tools = [retriever_tool]

# Lines 101-103 add the ToolNode to the graph

workflow.add_node("tool_retrieve", ToolNode(tools))

Configure Conditional Routing with tools_condition

After adding the node, establish edges that route execution to the tool when tool_calls are detected. The tools_condition function from LangGraph inspects the state messages for tool call requests.

# src/services/agents/agentic_rag.py

# Lines 26-34 configure conditional routing

workflow.add_conditional_edges(
    "retrieve",
    tools_condition,                # Inspects state["messages"] for tool_calls

    {
        "tools": "tool_retrieve",   # Routes to ToolNode when tools are requested

        END: END,
    },
)

tools_condition automatically detects when a node emits an AIMessage containing tool_calls and routes to the configured tool node. When the tool completes, execution returns to the calling node or proceeds to the next step in the workflow.

Step 3: Trigger the Tool from Agent Nodes

Agent nodes initiate tool execution by emitting an AIMessage with a populated tool_calls list. In src/services/agents/nodes/retrieve_node.py, the retrieve node constructs this message to trigger the retriever tool.

# src/services/agents/nodes/retrieve_node.py

# Lines 92-101 emit the tool call

updates["messages"] = [
    AIMessage(
        content="",
        tool_calls=[
            {
                "id": f"retrieve_{new_attempt_count}",
                "name": "retrieve_papers",   # Must match the @tool function name

                "args": {"query": question},
            }
        ],
    )
]

When LangGraph processes this state update, tools_condition detects the tool_calls entry, routes execution to the tool_retrieve node, and invokes retrieve_papers with the provided arguments. The tool's return value flows back into the agent state through extract_tool_artefacts in nodes/utils.py, making the results available for downstream grading and answer generation.

Practical Example: Adding a Calculator Tool

To add a math calculator alongside the existing retriever, follow the same three-step pattern.

1. Define the tool in tools.py:

# src/services/agents/tools.py

from langchain_core.tools import tool

def create_calculator_tool():
    @tool
    async def calc(expression: str) -> str:
        """Evaluate a simple arithmetic expression safely."""
        allowed = "0123456789.+-*/() "
        if any(ch not in allowed for ch in expression):
            raise ValueError("Invalid characters in expression")
        try:
            result = eval(expression, {"__builtins__": {}})
        except Exception as e:
            raise ValueError(f"Failed to evaluate: {e}")
        return str(result)
    return calc

2. Register in _build_graph:

# src/services/agents/agentic_rag.py

calculator_tool = create_calculator_tool()
tools = [retriever_tool, calculator_tool]  # Extend the list

workflow.add_node("tool_retrieve", ToolNode(tools))

3. Trigger from a node:

# Example payload emitted by any agent node

AIMessage(
    content="",
    tool_calls=[
        {
            "id": "calc_1",
            "name": "calc",               # Matches the @tool function name

            "args": {"expression": "12 * (7 - 3)"},
        }
    ],
)

The calculator executes within the same ToolNode, and its string result stores in relevant_tool_artefacts for later use in answer generation or reasoning steps.

Managing Tool Results in Agent State

The workflow persists tool outputs in src/services/agents/state.py, specifically within the relevant_tool_artefacts field defined at lines 41-50.

# src/services/agents/state.py

relevant_tool_artefacts: Optional[List[ToolArtefact]]

The ToolArtefact model in src/services/agents/models.py captures the tool name, call ID, content, and metadata, enabling downstream nodes to correlate results with specific tool invocations during multi-step reasoning.

Summary

• Define tools using the @tool decorator from langchain_core.tools in src/services/agents/tools.py, using async factories for dependency injection.
• Register tools in AgenticRAGService._build_graph by adding them to the ToolNode and configuring tools_condition for routing.
• Trigger tools by emitting tool_calls entries in AIMessage objects from any agent node, ensuring the "name" matches the decorated function name.
• Access results through the relevant_tool_artefacts field in AgentState, which stores serialized tool outputs for downstream processing.

Frequently Asked Questions

Can I add synchronous tools to a LangGraph agent workflow?

While LangGraph supports synchronous functions, the jamwithai/production-agentic-rag-course implementation uses async patterns throughout for non-blocking I/O. If you must use a synchronous tool, wrap it in asyncio.to_thread or define it as def instead of async def, but prefer async for consistency with opensearch and embedding clients.

How does LangGraph route execution to the correct tool when multiple tools are registered?

The tools_condition function inspects the tool_calls list in the state messages and routes to the single ToolNode containing all registered tools. The ToolNode then dispatches to the specific tool based on the "name" field in each tool_calls entry. You do not need separate routing logic per tool.

What happens if a tool raises an exception during execution?

LangGraph's ToolNode catches exceptions and returns them as ToolMessage objects with error status, which flow back into the agent state. Your agent nodes should inspect relevant_tool_artefacts or the message history to detect failures and implement retry logic or error handling as demonstrated in the retrieve node's retry patterns.

Where should I initialize expensive resources like database connections for custom tools?

Follow the factory pattern shown in create_retriever_tool: accept initialized clients (OpenSearch, embeddings) as factory arguments rather than constructing them inside the tool function. Instantiate these clients in AgenticRAGService.__init__ and pass them to the factory during _build_graph, ensuring connections are reused across tool invocations and properly managed with the service lifecycle.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}