LangGraph ToolNode & tool_calls Format Errors

from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import BaseTool


def build_custom_tool_node(tools: list[BaseTool]):
    """Roll your own ToolNode — useful when you need custom error handling."""
    tools_by_name = {t.name: t for t in tools}

    def tool_node(state):
        last = state["messages"][-1]
        if not isinstance(last, AIMessage) or not last.tool_calls:
            return {}   # nothing to do

        results = []
        for call in last.tool_calls:
            tool_name = call["name"]
            tool_id = call["id"]              # <-- REQUIRED — pair with the call
            tool_args = call["args"]

            if tool_name not in tools_by_name:
                # Emit an error ToolMessage — do NOT omit the response entirely,
                # or the model provider will reject the next turn.
                results.append(ToolMessage(
                    content=f"Error: tool '{tool_name}' is not available.",
                    tool_call_id=tool_id,
                    status="error",
                ))
                continue

            try:
                output = tools_by_name[tool_name].invoke(tool_args)
                results.append(ToolMessage(
                    content=str(output),
                    tool_call_id=tool_id,     # <-- match by id
                ))
            except Exception as e:
                # Always emit a ToolMessage even on failure
                results.append(ToolMessage(
                    content=f"Tool raised: {type(e).__name__}: {e}",
                    tool_call_id=tool_id,
                    status="error",
                ))

        return {"messages": results}

    return tool_node


# ✅ Wire into a graph
from langgraph.graph import StateGraph, START, END, MessagesState

def call_model(state):
    resp = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": [resp]}


def route(state):
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return END


graph = StateGraph(MessagesState)
graph.add_node("model", call_model)
graph.add_node("tools", build_custom_tool_node([search, calculator]))
graph.add_edge(START, "model")
graph.add_conditional_edges("model", route)
graph.add_edge("tools", "model")   # back to model with tool results
app = graph.compile(checkpointer=memory)

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool


@tool
def divide(a: float, b: float) -> float:
    """Divide a by b."""
    return a / b       # ZeroDivisionError if b == 0


@tool
def flaky_search(query: str) -> str:
    """Search that sometimes fails (simulating a real external service)."""
    import random
    if random.random() < 0.1:
        raise ConnectionError("Search service unavailable")
    return f"results for {query}"


# ✅ Default — errors become ToolMessage(status="error"), graph continues
tools = [divide, flaky_search]
tool_node = ToolNode(tools, handle_tool_errors=True)


# ✅ Custom error formatter — full control over what the model sees on failure
def format_error(exc: Exception) -> str:
    if isinstance(exc, ZeroDivisionError):
        return "Cannot divide by zero. Ask the user to provide a nonzero divisor."
    if isinstance(exc, ConnectionError):
        return f"Search is temporarily unavailable ({exc}). You can try answering from prior context or ask the user to retry."
    return f"Tool failed: {type(exc).__name__}: {exc}"

tool_node = ToolNode(tools, handle_tool_errors=format_error)


# ✅ Different handling per exception type — pass a callable that raises what you want to abort on
def handler(exc: Exception) -> str:
    if isinstance(exc, KeyboardInterrupt):
        raise                              # let interrupt propagate
    return f"Tool error: {exc}"

tool_node = ToolNode(tools, handle_tool_errors=handler)


# ✅ Add per-call logging with a subclass
class LoggingToolNode(ToolNode):
    def __init__(self, tools, **kw):
        super().__init__(tools, **kw)

    def _run_one(self, call, *args, **kw):
        import time, logging
        t0 = time.perf_counter()
        try:
            result = super()._run_one(call, *args, **kw)
            logging.info(
                "tool_call=%s ms=%d ok",
                call["name"], int((time.perf_counter() - t0) * 1000),
            )
            return result
        except Exception as e:
            logging.warning(
                "tool_call=%s ms=%d err=%s",
                call["name"], int((time.perf_counter() - t0) * 1000), e,
            )
            raise


# NOTE: private-method override is version-fragile — pin your LangGraph version if you rely on it.
tool_node = LoggingToolNode(tools, handle_tool_errors=True)

from langchain_core.tools import tool
import textwrap
import json


# ❌ BUG — search dumps raw HTML, easily 50-500KB per call
@tool
def bad_search(query: str) -> str:
    """Search the web and return the raw HTML of the top result."""
    import requests
    r = requests.get(f"https://api.search.example/?q={query}")
    return r.text     # <-- could be huge


# ✅ Fix — bound the output; summarize + link
MAX_TOOL_OUTPUT_CHARS = 8_000

@tool
def search(query: str, top_k: int = 5) -> str:
    """Search the web. Returns up to top_k concise result summaries.
    Use `fetch_url` to read the full content of any result URL.
    """
    import requests
    r = requests.get(
        f"https://api.search.example/", params={"q": query, "n": top_k}
    ).json()
    lines = []
    for hit in r["results"][:top_k]:
        lines.append(f"- [{hit['title']}]({hit['url']})\n  {hit['snippet'][:200]}")
    result = "\n".join(lines)
    return _truncate(result)


@tool
def fetch_url(url: str, max_chars: int = 8_000) -> str:
    """Fetch and return the readable text of a URL. Truncates to max_chars."""
    import requests
    from bs4 import BeautifulSoup
    r = requests.get(url, timeout=10)
    text = BeautifulSoup(r.text, "html.parser").get_text(" ", strip=True)
    return _truncate(text, max_chars)


@tool
def query_db(sql: str) -> str:
    """Run a SQL SELECT and return results as JSON. Caps rows at 50."""
    rows = db.execute(sql).fetchmany(50)
    payload = {
        "rows": [dict(r) for r in rows],
        "truncated": len(rows) >= 50,
    }
    return _truncate(json.dumps(payload, default=str))


def _truncate(s: str, cap: int = MAX_TOOL_OUTPUT_CHARS) -> str:
    if len(s) <= cap:
        return s
    return s[:cap] + f"\n\n[... truncated {len(s) - cap} chars. Refine your query for more specific results.]"


# ✅ At the ToolNode level — hard cap on any tool output
from langgraph.prebuilt import ToolNode
from langchain_core.messages import ToolMessage

class CappedToolNode(ToolNode):
    def _run_one(self, call, *args, **kw):
        msg = super()._run_one(call, *args, **kw)
        if isinstance(msg, ToolMessage) and len(msg.content) > MAX_TOOL_OUTPUT_CHARS:
            msg.content = _truncate(msg.content)
        return msg

tool_node = CappedToolNode([search, fetch_url, query_db])