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

> Learn to add custom tools to your LangGraph agent workflow. This guide covers defining functions, registering them in ToolNode, and handling tool calls for seamless integration.

- Repository: [jamwithai/production-agentic-rag-course](https://github.com/jamwithai/production-agentic-rag-course)
- Tags: how-to-guide
- Published: 2026-03-23

---

**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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`.

```python

# 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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/agents/agentic_rag.py)** within the `AgenticRAGService._build_graph` method. Here you instantiate the tool and add it to the `ToolNode`.

```python

# 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.

```python

# 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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/agents/nodes/retrieve_node.py)**, the retrieve node constructs this message to trigger the retriever tool.

```python

# 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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/tools.py):**

```python

# 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`:**

```python

# 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:**

```python

# 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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/agents/state.py)**, specifically within the `relevant_tool_artefacts` field defined at lines 41-50.

```python

# src/services/agents/state.py

relevant_tool_artefacts: Optional[List[ToolArtefact]]

```

The **`ToolArtefact`** model in [`src/services/agents/models.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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.