codekingpro/portable-devtools
115k
1"""Tool execution node for LangGraph workflows.2 3This module provides prebuilt functionality for executing tools in LangGraph.4 5Tools are functions that models can call to interact with external systems,6APIs, databases, or perform computations.7 8The module implements design patterns for:9 10- Parallel execution of multiple tool calls for efficiency11- Robust error handling with customizable error messages12- State injection for tools that need access to graph state13- Store injection for tools that need persistent storage14- Command-based state updates for advanced control flow15 16Key Components:17 18- [`ToolNode`][langgraph.prebuilt.ToolNode]: Main class for executing tools in LangGraph workflows19- [`InjectedState`][langgraph.prebuilt.InjectedState]: Annotation for injecting graph state into tools20- [`InjectedStore`][langgraph.prebuilt.InjectedStore]: Annotation for injecting persistent store into tools21- [`ToolRuntime`][langgraph.prebuilt.ToolRuntime]: Runtime information for tools, bundling together `state`, `context`,22 `config`, `stream_writer`, `tool_call_id`, and `store`23- [`tools_condition`][langgraph.prebuilt.tools_condition]: Utility function for conditional routing based on tool calls24 25Typical Usage:26 ```python27 from langchain_core.tools import tool28 from langchain.tools import ToolNode29 30 31 @tool32 def my_tool(x: int) -> str:33 return f"Result: {x}"34 35 36 tool_node = ToolNode([my_tool])37 ```38"""39 40from __future__ import annotations41 42import asyncio43import inspect44import json45from collections.abc import Awaitable, Callable46from copy import copy, deepcopy47from dataclasses import dataclass, field, replace48from types import UnionType49from typing import (50 TYPE_CHECKING,51 Annotated,52 Any,53 Generic,54 Literal,55 TypedDict,56 Union,57 cast,58 get_args,59 get_origin,60 get_type_hints,61)62 63from langchain_core.messages import (64 AIMessage,65 AnyMessage,66 RemoveMessage,67 ToolCall,68 ToolMessage,69 convert_to_messages,70)71from langchain_core.runnables.config import (72 RunnableConfig,73 get_config_list,74 get_executor_for_config,75)76from langchain_core.tools import BaseTool, InjectedToolArg77from langchain_core.tools import tool as create_tool78from langchain_core.tools.base import (79 TOOL_MESSAGE_BLOCK_TYPES,80 ToolException,81 _DirectlyInjectedToolArg,82 _is_injected_arg_type,83 get_all_basemodel_annotations,84)85from langgraph._internal._constants import CONF, CONFIG_KEY_READ86from langgraph._internal._runnable import RunnableCallable87from langgraph.errors import GraphBubbleUp88from langgraph.graph.message import REMOVE_ALL_MESSAGES89from langgraph.pregel._tools import _tool_call_writer90from langgraph.runtime import ExecutionInfo, ServerInfo # noqa: TC00291from langgraph.store.base import BaseStore # noqa: TC00292from langgraph.types import Command, Send, StreamWriter93from pydantic import BaseModel, ValidationError94from typing_extensions import TypeVar, Unpack95 96if TYPE_CHECKING:97 from collections.abc import Sequence98 99 from langgraph.runtime import Runtime100 from pydantic_core import ErrorDetails101 102# right now we use a dict as the default, can change this to AgentState, but depends103# on if this lives in LangChain or LangGraph... ideally would have some typed104# messages key105StateT = TypeVar("StateT", default=dict)106ContextT = TypeVar("ContextT", default=None)107 108INVALID_TOOL_NAME_ERROR_TEMPLATE = (109 "Error: {requested_tool} is not a valid tool, try one of [{available_tools}]."110)111TOOL_CALL_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."112TOOL_EXECUTION_ERROR_TEMPLATE = (113 "Error executing tool '{tool_name}' with kwargs {tool_kwargs} with error:\n"114 " {error}\n"115 " Please fix the error and try again."116)117TOOL_INVOCATION_ERROR_TEMPLATE = (118 "Error invoking tool '{tool_name}' with kwargs {tool_kwargs} with error:\n"119 " {error}\n"120 " Please fix the error and try again."121)122 123 124class _ToolCallRequestOverrides(TypedDict, total=False):125 """Possible overrides for ToolCallRequest.override() method."""126 127 tool_call: ToolCall128 tool: BaseTool129 state: Any130 131 132@dataclass133class ToolCallRequest:134 """Tool execution request passed to tool call interceptors.135 136 Attributes:137 tool_call: Tool call dict with name, args, and id from model output.138 tool: BaseTool instance to be invoked, or None if tool is not139 registered with the `ToolNode`. When tool is `None`, interceptors can140 handle the request without validation. If the interceptor calls `execute()`,141 validation will occur and raise an error for unregistered tools.142 state: Agent state (`dict`, `list`, or `BaseModel`).143 runtime: LangGraph runtime context (optional, `None` if outside graph).144 """145 146 tool_call: ToolCall147 tool: BaseTool | None148 state: Any149 runtime: ToolRuntime150 151 def __setattr__(self, name: str, value: Any) -> None:152 """Raise deprecation warning when setting attributes directly.153 154 Direct attribute assignment is deprecated. Use the `override()` method instead.155 """156 import warnings157 158 # Allow setting attributes during initialization159 if not hasattr(self, "__dataclass_fields__") or not hasattr(self, name):160 object.__setattr__(self, name, value)161 else:162 warnings.warn(163 f"Setting attribute '{name}' on ToolCallRequest is deprecated. "164 "Use the override() method instead to create a new instance with modified values.",165 DeprecationWarning,166 stacklevel=2,167 )168 object.__setattr__(self, name, value)169 170 def override(171 self, **overrides: Unpack[_ToolCallRequestOverrides]172 ) -> ToolCallRequest:173 """Replace the request with a new request with the given overrides.174 175 Returns a new `ToolCallRequest` instance with the specified attributes replaced.176 This follows an immutable pattern, leaving the original request unchanged.177 178 Args:179 **overrides: Keyword arguments for attributes to override.180 181 Supported keys:182 183 - tool_call: Tool call dict with `name`, `args`, and `id`184 - state: Agent state (`dict`, `list`, or `BaseModel`)185 186 Returns:187 New ToolCallRequest instance with specified overrides applied.188 189 Examples:190 ```python191 # Modify tool call arguments without mutating original192 modified_call = {**request.tool_call, "args": {"value": 10}}193 new_request = request.override(tool_call=modified_call)194 195 # Override multiple attributes196 new_request = request.override(tool_call=modified_call, state=new_state)197 ```198 """199 return replace(self, **overrides)200 201 202ToolCallWrapper = Callable[203 [ToolCallRequest, Callable[[ToolCallRequest], ToolMessage | Command]],204 ToolMessage | Command,205]206"""Wrapper for tool call execution with multi-call support.207 208Wrapper receives:209 request: ToolCallRequest with tool_call, tool, state, and runtime.210 execute: Callable to execute the tool (CAN BE CALLED MULTIPLE TIMES).211 212Returns:213 ToolMessage or Command (the final result).214 215The execute callable can be invoked multiple times for retry logic,216with potentially modified requests each time. Each call to execute217is independent and stateless.218 219!!! note220 When implementing middleware for `create_agent`, use221 `AgentMiddleware.wrap_tool_call` which provides properly typed222 state parameter for better type safety.223 224Examples:225 Passthrough (execute once):226 227 def handler(request, execute):228 return execute(request)229 230 Modify request before execution:231 232 ```python233 def handler(request, execute):234 modified_call = {**request.tool_call, "args": {**request.tool_call["args"], "value": request.tool_call["args"]["value"] * 2}}235 modified_request = request.override(tool_call=modified_call)236 return execute(modified_request)237 ```238 239 Retry on error (execute multiple times):240 241 ```python242 def handler(request, execute):243 for attempt in range(3):244 try:245 result = execute(request)246 if is_valid(result):247 return result248 except Exception:249 if attempt == 2:250 raise251 return result252 ```253 254 Conditional retry based on response:255 256 ```python257 def handler(request, execute):258 for attempt in range(3):259 result = execute(request)260 if isinstance(result, ToolMessage) and result.status != "error":261 return result262 if attempt < 2:263 continue264 return result265 ```266 267 Cache/short-circuit without calling execute:268 269 ```python270 def handler(request, execute):271 if cached := get_cache(request):272 return ToolMessage(content=cached, tool_call_id=request.tool_call["id"])273 result = execute(request)274 save_cache(request, result)275 return result276 ```277"""278 279AsyncToolCallWrapper = Callable[280 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]]],281 Awaitable[ToolMessage | Command],282]283"""Async wrapper for tool call execution with multi-call support."""284 285 286class ToolCallWithContext(TypedDict):287 """ToolCall with additional context for graph state.288 289 This is an internal data structure meant to help the `ToolNode` accept290 tool calls with additional context (e.g. state) when dispatched using the291 Send API.292 293 The Send API is used in create_agent to distribute tool calls in parallel294 and support human-in-the-loop workflows where graph execution may be paused295 for an indefinite time.296 """297 298 tool_call: ToolCall299 __type: Literal["tool_call_with_context"]300 """Type to parameterize the payload.301 302 Using "__" as a prefix to be defensive against potential name collisions with303 regular user state.304 """305 state: Any306 """The state is provided as additional context."""307 308 309def msg_content_output(output: Any) -> str | list[dict]:310 """Convert tool output to `ToolMessage` content format.311 312 Handles `str`, `list[dict]` (content blocks), and arbitrary objects by attempting313 JSON serialization with fallback to str().314 315 Args:316 output: Tool execution output of any type.317 318 Returns:319 String or list of content blocks suitable for `ToolMessage.content`.320 """321 if isinstance(output, str) or (322 isinstance(output, list)323 and all(324 isinstance(x, dict) and x.get("type") in TOOL_MESSAGE_BLOCK_TYPES325 for x in output326 )327 ):328 return output329 # Technically a list of strings is also valid message content, but it's330 # not currently well tested that all chat models support this.331 # And for backwards compatibility we want to make sure we don't break332 # any existing ToolNode usage.333 try:334 return json.dumps(output, ensure_ascii=False)335 except Exception: # noqa: BLE001336 return str(output)337 338 339class ToolInvocationError(ToolException):340 """An error occurred while invoking a tool due to invalid arguments.341 342 This exception is only raised when invoking a tool using the `ToolNode`!343 """344 345 def __init__(346 self,347 tool_name: str,348 source: ValidationError,349 tool_kwargs: dict[str, Any],350 filtered_errors: list[ErrorDetails] | None = None,351 ) -> None:352 """Initialize the ToolInvocationError.353 354 Args:355 tool_name: The name of the tool that failed.356 source: The exception that occurred.357 tool_kwargs: The keyword arguments that were passed to the tool.358 filtered_errors: Optional list of filtered validation errors excluding359 injected arguments.360 """361 # Format error display based on filtered errors if provided362 if filtered_errors is not None:363 # Manually format the filtered errors without URLs or fancy formatting364 error_str_parts = []365 for error in filtered_errors:366 loc_str = ".".join(str(loc) for loc in error.get("loc", ()))367 msg = error.get("msg", "Unknown error")368 error_str_parts.append(f"{loc_str}: {msg}")369 error_display_str = "\n".join(error_str_parts)370 else:371 error_display_str = str(source)372 373 self.message = TOOL_INVOCATION_ERROR_TEMPLATE.format(374 tool_name=tool_name, tool_kwargs=tool_kwargs, error=error_display_str375 )376 self.tool_name = tool_name377 self.tool_kwargs = tool_kwargs378 self.source = source379 self.filtered_errors = filtered_errors380 super().__init__(self.message)381 382 383def _default_handle_tool_errors(e: Exception) -> str:384 """Default error handler for tool errors.385 386 If the tool is a tool invocation error, return its message.387 Otherwise, raise the error.388 """389 if isinstance(e, ToolInvocationError):390 return e.message391 raise e392 393 394def _handle_tool_error(395 e: Exception,396 *,397 flag: bool398 | str399 | Callable[..., str]400 | type[Exception]401 | tuple[type[Exception], ...],402) -> str:403 """Generate error message content based on exception handling configuration.404 405 This function centralizes error message generation logic, supporting different406 error handling strategies configured via the `ToolNode`'s `handle_tool_errors`407 parameter.408 409 Args:410 e: The exception that occurred during tool execution.411 flag: Configuration for how to handle the error. Can be:412 - bool: If `True`, use default error template413 - str: Use this string as the error message414 - Callable: Call this function with the exception to get error message415 - tuple: Not used in this context (handled by caller)416 417 Returns:418 A string containing the error message to include in the `ToolMessage`.419 420 Raises:421 ValueError: If flag is not one of the supported types.422 423 !!! note424 The tuple case is handled by the caller through exception type checking,425 not by this function directly.426 """427 if isinstance(flag, (bool, tuple)) or (428 isinstance(flag, type) and issubclass(flag, Exception)429 ):430 content = TOOL_CALL_ERROR_TEMPLATE.format(error=repr(e))431 elif isinstance(flag, str):432 content = flag433 elif callable(flag):434 content = flag(e) # type: ignore [assignment, call-arg]435 else:436 msg = (437 f"Got unexpected type of `handle_tool_error`. Expected bool, str "438 f"or callable. Received: {flag}"439 )440 raise ValueError(msg)441 return content442 443 444def _infer_handled_types(handler: Callable[..., str]) -> tuple[type[Exception], ...]:445 """Infer exception types handled by a custom error handler function.446 447 This function analyzes the type annotations of a custom error handler to determine448 which exception types it's designed to handle. This enables type-safe error handling449 where only specific exceptions are caught and processed by the handler.450 451 Args:452 handler: A callable that takes an exception and returns an error message string.453 The first parameter (after self/cls if present) should be type-annotated454 with the exception type(s) to handle.455 456 Returns:457 A tuple of exception types that the handler can process. Returns (Exception,)458 if no specific type information is available for backward compatibility.459 460 Raises:461 ValueError: If the handler's annotation contains non-Exception types or462 if Union types contain non-Exception types.463 464 !!! note465 This function supports both single exception types and Union types for466 handlers that need to handle multiple exception types differently.467 """468 sig = inspect.signature(handler)469 params = list(sig.parameters.values())470 if params:471 # If it's a method, the first argument is typically 'self' or 'cls'472 if params[0].name in ["self", "cls"] and len(params) == 2:473 first_param = params[1]474 else:475 first_param = params[0]476 477 type_hints = get_type_hints(handler)478 if first_param.name in type_hints:479 origin = get_origin(first_param.annotation)480 if origin in [Union, UnionType]:481 args = get_args(first_param.annotation)482 if all(issubclass(arg, Exception) for arg in args):483 return tuple(args)484 msg = (485 "All types in the error handler error annotation must be "486 "Exception types. For example, "487 "`def custom_handler(e: Union[ValueError, TypeError])`. "488 f"Got '{first_param.annotation}' instead."489 )490 raise ValueError(msg)491 492 exception_type = type_hints[first_param.name]493 if Exception in exception_type.__mro__:494 return (exception_type,)495 msg = (496 f"Arbitrary types are not supported in the error handler "497 f"signature. Please annotate the error with either a "498 f"specific Exception type or a union of Exception types. "499 "For example, `def custom_handler(e: ValueError)` or "500 "`def custom_handler(e: Union[ValueError, TypeError])`. "501 f"Got '{exception_type}' instead."502 )503 raise ValueError(msg)504 505 # If no type information is available, return (Exception,)506 # for backwards compatibility.507 return (Exception,)508 509 510def _filter_validation_errors(511 validation_error: ValidationError,512 injected_args: _InjectedArgs | None,513) -> list[ErrorDetails]:514 """Filter validation errors to only include LLM-controlled arguments.515 516 When a tool invocation fails validation, only errors for arguments that the LLM517 controls should be included in error messages. This ensures the LLM receives518 focused, actionable feedback about parameters it can actually fix. System-injected519 arguments (state, store, runtime) are filtered out since the LLM has no control520 over them.521 522 This function also removes injected argument values from the `input` field in error523 details, ensuring that only LLM-provided arguments appear in error messages.524 525 Args:526 validation_error: The Pydantic ValidationError raised during tool invocation.527 injected_args: The _InjectedArgs structure containing all injected arguments,528 or None if there are no injected arguments.529 530 Returns:531 List of ErrorDetails containing only errors for LLM-controlled arguments,532 with system-injected argument values removed from the input field.533 """534 # Collect all injected argument names535 injected_arg_names: set[str] = set()536 if injected_args:537 if injected_args.state:538 injected_arg_names.update(injected_args.state.keys())539 if injected_args.store:540 injected_arg_names.add(injected_args.store)541 if injected_args.runtime:542 injected_arg_names.add(injected_args.runtime)543 544 filtered_errors: list[ErrorDetails] = []545 for error in validation_error.errors():546 # Check if error location contains any injected argument547 # error['loc'] is a tuple like ('field_name',) or ('field_name', 'nested_field')548 if error["loc"] and error["loc"][0] not in injected_arg_names:549 # Create a copy of the error dict to avoid mutating the original550 error_copy: dict[str, Any] = {**error}551 552 # Remove injected arguments from input_value if it's a dict553 if isinstance(error_copy.get("input"), dict):554 input_dict = error_copy["input"]555 input_copy = {556 k: v for k, v in input_dict.items() if k not in injected_arg_names557 }558 error_copy["input"] = input_copy559 560 # Cast is safe because ErrorDetails is a TypedDict compatible with this structure561 filtered_errors.append(error_copy) # type: ignore[arg-type]562 563 return filtered_errors564 565 566@dataclass567class _InjectedArgs:568 """Internal structure for tracking injected arguments for a tool.569 570 This data structure is built once during ToolNode initialization by analyzing571 the tool's signature and args schema, then reused during execution for efficient572 injection without repeated reflection.573 574 The structure maps from tool parameter names to their injection sources, enabling575 the ToolNode to know exactly which arguments need to be injected and where to576 get their values from.577 578 Attributes:579 state: Mapping from tool parameter names to state field names for injection.580 Keys are tool parameter names, values are either:581 - str: Name of the state field to extract and inject582 - None: Inject the entire state object583 Empty dict if no state injection is needed.584 store: Name of the tool parameter where the store should be injected,585 or None if no store injection is needed.586 runtime: Name of the tool parameter where the runtime should be injected,587 or None if no runtime injection is needed.588 589 Example:590 For a tool with signature:591 ```python592 def my_tool(593 x: int,594 messages: Annotated[list, InjectedState("messages")],595 full_state: Annotated[dict, InjectedState()],596 store: Annotated[BaseStore, InjectedStore()],597 runtime: ToolRuntime,598 ) -> str:599 ...600 ```601 602 The resulting `_InjectedArgs` would be:603 ```python604 _InjectedArgs(605 state={606 "messages": "messages", # Extract state["messages"]607 "full_state": None, # Inject entire state608 },609 store="store", # Inject into "store" parameter610 runtime="runtime", # Inject into "runtime" parameter611 )612 ```613 """614 615 state: dict[str, str | None]616 store: str | None617 runtime: str | None618 all_injected_keys: set[str]619 _optional_state_args: set[str]620 621 622class ToolNode(RunnableCallable):623 """A node for executing tools in LangGraph workflows.624 625 Handles tool execution patterns including function calls, state injection,626 persistent storage, and control flow. Manages parallel execution,627 error handling.628 629 Use `ToolNode` when building custom workflows that require fine-grained control over630 tool execution—for example, custom routing logic, specialized error handling, or631 non-standard agent architectures.632 633 For standard ReAct-style agents, use [`create_agent`][langchain.agents.create_agent]634 instead. It uses `ToolNode` internally with sensible defaults for the agent loop,635 conditional routing, and error handling.636 637 Input Formats:638 1. **Graph state** with `messages` key that has a list of messages:639 - Common representation for agentic workflows640 - Supports custom messages key via `messages_key` parameter641 642 2. **Message List**: `[AIMessage(..., tool_calls=[...])]`643 - List of messages with tool calls in the last AIMessage644 645 3. **Direct Tool Calls**: `[{"name": "tool", "args": {...}, "id": "1", "type": "tool_call"}]`646 - Bypasses message parsing for direct tool execution647 - For programmatic tool invocation and testing648 649 Output Formats:650 Output format depends on input type and tool behavior:651 652 **For Regular tools**:653 654 - Dict input → `{"messages": [ToolMessage(...)]}`655 - List input → `[ToolMessage(...)]`656 657 **For Command tools**:658 659 - Returns `[Command(...)]` or mixed list with regular tool outputs660 - `Command` can update state, trigger navigation, or send messages661 662 Args:663 tools: A sequence of tools that can be invoked by this node.664 665 Supports:666 667 - **BaseTool instances**: Tools with schemas and metadata668 - **Plain functions**: Automatically converted to tools with inferred schemas669 670 name: The name identifier for this node in the graph. Used for debugging671 and visualization.672 tags: Optional metadata tags to associate with the node for filtering673 and organization.674 handle_tool_errors: Configuration for error handling during tool execution.675 Supports multiple strategies:676 677 - `True`: Catch all errors and return a `ToolMessage` with the default678 error template containing the exception details.679 - `str`: Catch all errors and return a `ToolMessage` with this custom680 error message string.681 - `type[Exception]`: Only catch exceptions with the specified type and682 return the default error message for it.683 - `tuple[type[Exception], ...]`: Only catch exceptions with the specified684 types and return default error messages for them.685 - `Callable[..., str]`: Catch exceptions matching the callable's signature686 and return the string result of calling it with the exception.687 - `False`: Disable error handling entirely, allowing exceptions to688 propagate.689 690 Defaults to a callable that:691 692 - Catches tool invocation errors (due to invalid arguments provided by the693 model) and returns a descriptive error message694 - Ignores tool execution errors (they will be re-raised)695 696 messages_key: The key in the state dictionary that contains the message list.697 This same key will be used for the output `ToolMessage` objects.698 699 Allows custom state schemas with different message field names.700 701 Examples:702 Basic usage:703 704 ```python705 from langchain.tools import ToolNode706 from langchain_core.tools import tool707 708 @tool709 def calculator(a: int, b: int) -> int:710 \"\"\"Add two numbers.\"\"\"711 return a + b712 713 tool_node = ToolNode([calculator])714 ```715 716 State injection:717 718 ```python719 from typing_extensions import Annotated720 from langchain.tools import InjectedState721 722 @tool723 def context_tool(query: str, state: Annotated[dict, InjectedState]) -> str:724 \"\"\"Some tool that uses state.\"\"\"725 return f"Query: {query}, Messages: {len(state['messages'])}"726 727 tool_node = ToolNode([context_tool])728 ```729 730 Error handling:731 732 ```python733 def handle_errors(e: ValueError) -> str:734 return "Invalid input provided"735 736 737 tool_node = ToolNode([my_tool], handle_tool_errors=handle_errors)738 ```739 """ # noqa: E501740 741 name: str = "tools"742 743 def __init__(744 self,745 tools: Sequence[BaseTool | Callable],746 *,747 name: str = "tools",748 tags: list[str] | None = None,749 handle_tool_errors: bool750 | str751 | Callable[..., str]752 | type[Exception]753 | tuple[type[Exception], ...] = _default_handle_tool_errors,754 messages_key: str = "messages",755 wrap_tool_call: ToolCallWrapper | None = None,756 awrap_tool_call: AsyncToolCallWrapper | None = None,757 ) -> None:758 """Initialize `ToolNode` with tools and configuration.759 760 Args:761 tools: Sequence of tools to make available for execution.762 name: Node name for graph identification.763 tags: Optional metadata tags.764 handle_tool_errors: Error handling configuration.765 messages_key: State key containing messages.766 wrap_tool_call: Sync wrapper function to intercept tool execution. Receives767 ToolCallRequest and execute callable, returns ToolMessage or Command.768 Enables retries, caching, request modification, and control flow.769 awrap_tool_call: Async wrapper function to intercept tool execution.770 If not provided, falls back to wrap_tool_call for async execution.771 """772 super().__init__(self._func, self._afunc, name=name, tags=tags, trace=False)773 self._tools_by_name: dict[str, BaseTool] = {}774 self._injected_args: dict[str, _InjectedArgs] = {}775 self._handle_tool_errors = handle_tool_errors776 self._messages_key = messages_key777 self._wrap_tool_call = wrap_tool_call778 self._awrap_tool_call = awrap_tool_call779 for tool in tools:780 if not isinstance(tool, BaseTool):781 tool_ = create_tool(cast("type[BaseTool]", tool))782 else:783 tool_ = tool784 self._tools_by_name[tool_.name] = tool_785 # Build injected args mapping once during initialization in a single pass786 self._injected_args[tool_.name] = _get_all_injected_args(tool_)787 788 @property789 def tools_by_name(self) -> dict[str, BaseTool]:790 """Mapping from tool name to BaseTool instance."""791 return self._tools_by_name792 793 def _func(794 self,795 input: list[AnyMessage] | dict[str, Any] | BaseModel,796 config: RunnableConfig,797 runtime: Runtime,798 ) -> Any:799 tool_calls, input_type = self._parse_input(input)800 config_list = get_config_list(config, len(tool_calls))801 802 # Construct ToolRuntime instances at the top level for each tool call803 tool_runtimes = []804 for call, cfg in zip(tool_calls, config_list, strict=False):805 state = self._extract_state(input, cfg)806 tool_runtime = ToolRuntime(807 state=state,808 tool_call_id=call["id"],809 config=cfg,810 context=runtime.context,811 store=runtime.store,812 stream_writer=runtime.stream_writer,813 tools=list(self.tools_by_name.values()),814 execution_info=runtime.execution_info,815 server_info=runtime.server_info,816 )817 tool_runtimes.append(tool_runtime)818 819 # Pass original tool calls without injection820 input_types = [input_type] * len(tool_calls)821 with get_executor_for_config(config) as executor:822 outputs = list(823 executor.map(self._run_one, tool_calls, input_types, tool_runtimes)824 )825 826 return self._combine_tool_outputs(outputs, input_type)827 828 async def _afunc(829 self,830 input: list[AnyMessage] | dict[str, Any] | BaseModel,831 config: RunnableConfig,832 runtime: Runtime,833 ) -> Any:834 tool_calls, input_type = self._parse_input(input)835 config_list = get_config_list(config, len(tool_calls))836 837 # Construct ToolRuntime instances at the top level for each tool call838 tool_runtimes = []839 for call, cfg in zip(tool_calls, config_list, strict=False):840 state = self._extract_state(input, cfg)841 tool_runtime = ToolRuntime(842 state=state,843 tool_call_id=call["id"],844 config=cfg,845 context=runtime.context,846 store=runtime.store,847 stream_writer=runtime.stream_writer,848 tools=list(self.tools_by_name.values()),849 execution_info=runtime.execution_info,850 server_info=runtime.server_info,851 )852 tool_runtimes.append(tool_runtime)853 854 # Pass original tool calls without injection855 coros = []856 for call, tool_runtime in zip(tool_calls, tool_runtimes, strict=False):857 coros.append(self._arun_one(call, input_type, tool_runtime)) # type: ignore[arg-type]858 outputs = await asyncio.gather(*coros)859 860 return self._combine_tool_outputs(outputs, input_type)861 862 def _combine_tool_outputs(863 self,864 outputs: list[ToolMessage | Command | list[ToolMessage | Command]],865 input_type: Literal["list", "dict", "tool_calls"],866 ) -> list[Command | list[ToolMessage] | dict[str, list[ToolMessage]]]:867 # Flatten list entries from tools that returned multiple items868 flat_outputs: list[ToolMessage | Command]869 if any(isinstance(output, list) for output in outputs):870 flat_outputs = []871 for output in outputs:872 if isinstance(output, list):873 flat_outputs.extend(output)874 else:875 flat_outputs.append(output)876 else:877 flat_outputs = cast("list[ToolMessage | Command]", outputs)878 879 # preserve existing behavior for non-command tool outputs for backwards880 # compatibility881 if not any(isinstance(output, Command) for output in flat_outputs):882 # TypedDict, pydantic, dataclass, etc. should all be able to load from dict883 return (884 flat_outputs885 if input_type == "list"886 else {self._messages_key: flat_outputs}887 )888 889 # LangGraph will automatically handle list of Command and non-command node890 # updates891 combined_outputs: list[892 Command | list[ToolMessage] | dict[str, list[ToolMessage]]893 ] = []894 895 # combine all parent commands with goto into a single parent command896 parent_command: Command | None = None897 for output in flat_outputs:898 if isinstance(output, Command):899 if (900 output.graph is Command.PARENT901 and isinstance(output.goto, list)902 and all(isinstance(send, Send) for send in output.goto)903 ):904 if parent_command:905 parent_command = replace(906 parent_command,907 goto=cast("list[Send]", parent_command.goto) + output.goto,908 )909 else:910 parent_command = Command(graph=Command.PARENT, goto=output.goto)911 else:912 combined_outputs.append(output)913 else:914 combined_outputs.append(915 [output] if input_type == "list" else {self._messages_key: [output]}916 )917 918 if parent_command:919 combined_outputs.append(parent_command)920 return combined_outputs921 922 def _execute_tool_sync(923 self,924 request: ToolCallRequest,925 input_type: Literal["list", "dict", "tool_calls"],926 config: RunnableConfig,927 ) -> ToolMessage | Command | list[Command | ToolMessage]:928 """Execute tool call with configured error handling.929 930 Args:931 request: Tool execution request.932 input_type: Input format.933 config: Runnable configuration.934 935 Returns:936 ToolMessage, Command, or list of Command/ToolMessage.937 938 Raises:939 Exception: If tool fails and handle_tool_errors is False.940 """941 call = request.tool_call942 tool = request.tool943 944 # Validate tool exists when we actually need to execute it945 if tool is None:946 if invalid_tool_message := self._validate_tool_call(call):947 return invalid_tool_message948 # This should never happen if validation works correctly949 msg = f"Tool {call['name']} is not registered with ToolNode"950 raise TypeError(msg)951 952 # Inject state, store, and runtime right before invocation953 injected_call = self._inject_tool_args(call, request.runtime, tool)954 call_args = {**injected_call, "type": "tool_call"}955 956 try:957 try:958 response = tool.invoke(call_args, config)959 except ValidationError as exc:960 # Filter out errors for injected arguments961 injected = self._injected_args.get(call["name"])962 filtered_errors = _filter_validation_errors(exc, injected)963 # Use original call["args"] without injected values for error reporting964 raise ToolInvocationError(965 call["name"], exc, call["args"], filtered_errors966 ) from exc967 968 # Inside try so validation errors route through _handle_tool_errors969 return self._normalize_tool_response(970 response, request.tool_call, input_type971 )972 973 # GraphInterrupt is a special exception that will always be raised.974 # It can be triggered in the following scenarios,975 # Where GraphInterrupt(GraphBubbleUp) is raised from an `interrupt` invocation976 # most commonly:977 # (1) a GraphInterrupt is raised inside a tool978 # (2) a GraphInterrupt is raised inside a graph node for a graph called as a tool979 # (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph980 # called as a tool981 # (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)982 except GraphBubbleUp:983 raise984 except Exception as e:985 # Determine which exception types are handled986 handled_types: tuple[type[Exception], ...]987 if isinstance(self._handle_tool_errors, type) and issubclass(988 self._handle_tool_errors, Exception989 ):990 handled_types = (self._handle_tool_errors,)991 elif isinstance(self._handle_tool_errors, tuple):992 handled_types = self._handle_tool_errors993 elif callable(self._handle_tool_errors) and not isinstance(994 self._handle_tool_errors, type995 ):996 handled_types = _infer_handled_types(self._handle_tool_errors)997 else:998 # default behavior is catching all exceptions999 handled_types = (Exception,)1000 1001 # Check if this error should be handled1002 if not self._handle_tool_errors or not isinstance(e, handled_types):1003 raise1004 1005 # Error is handled - create error ToolMessage1006 content = _handle_tool_error(e, flag=self._handle_tool_errors)1007 return ToolMessage(1008 content=content,1009 name=call["name"],1010 tool_call_id=call["id"],1011 status="error",1012 )1013 1014 def _run_one(1015 self,1016 call: ToolCall,1017 input_type: Literal["list", "dict", "tool_calls"],1018 tool_runtime: ToolRuntime,1019 ) -> ToolMessage | Command | list[Command | ToolMessage]:1020 """Execute single tool call with wrap_tool_call wrapper if configured.1021 1022 Args:1023 call: Tool call dict.1024 input_type: Input format.1025 tool_runtime: Tool runtime.1026 1027 Returns:1028 ToolMessage or Command.1029 """1030 # Validation is deferred to _execute_tool_sync to allow interceptors1031 # to short-circuit requests for unregistered tools1032 tool = self.tools_by_name.get(call["name"])1033 1034 # Create the tool request with state and runtime1035 tool_request = ToolCallRequest(1036 tool_call=call,1037 tool=tool,1038 state=tool_runtime.state,1039 runtime=tool_runtime,1040 )1041 1042 config = tool_runtime.config1043 1044 if self._wrap_tool_call is None:1045 # No wrapper - execute directly1046 return self._execute_tool_sync(tool_request, input_type, config)1047 1048 # Define execute callable that can be called multiple times1049 def execute(req: ToolCallRequest) -> ToolMessage | Command:1050 """Execute tool with given request. Can be called multiple times."""1051 return self._execute_tool_sync(req, input_type, config)1052 1053 # Call wrapper with request and execute callable1054 try:1055 return self._wrap_tool_call(tool_request, execute)1056 except Exception as e:1057 # Wrapper threw an exception1058 if not self._handle_tool_errors:1059 raise1060 # Convert to error message1061 content = _handle_tool_error(e, flag=self._handle_tool_errors)1062 return ToolMessage(1063 content=content,1064 name=tool_request.tool_call["name"],1065 tool_call_id=tool_request.tool_call["id"],1066 status="error",1067 )1068 1069 async def _execute_tool_async(1070 self,1071 request: ToolCallRequest,1072 input_type: Literal["list", "dict", "tool_calls"],1073 config: RunnableConfig,1074 ) -> ToolMessage | Command | list[Command | ToolMessage]:1075 """Execute tool call asynchronously with configured error handling.1076 1077 Args:1078 request: Tool execution request.1079 input_type: Input format.1080 config: Runnable configuration.1081 1082 Returns:1083 ToolMessage, Command, or list of Command/ToolMessage.1084 1085 Raises:1086 Exception: If tool fails and handle_tool_errors is False.1087 """1088 call = request.tool_call1089 tool = request.tool1090 1091 # Validate tool exists when we actually need to execute it1092 if tool is None:1093 if invalid_tool_message := self._validate_tool_call(call):1094 return invalid_tool_message1095 # This should never happen if validation works correctly1096 msg = f"Tool {call['name']} is not registered with ToolNode"1097 raise TypeError(msg)1098 1099 # Inject state, store, and runtime right before invocation1100 injected_call = self._inject_tool_args(call, request.runtime, tool)1101 call_args = {**injected_call, "type": "tool_call"}1102 1103 try:1104 try:1105 response = await tool.ainvoke(call_args, config)1106 except ValidationError as exc:1107 # Filter out errors for injected arguments1108 injected = self._injected_args.get(call["name"])1109 filtered_errors = _filter_validation_errors(exc, injected)1110 # Use original call["args"] without injected values for error reporting1111 raise ToolInvocationError(1112 call["name"], exc, call["args"], filtered_errors1113 ) from exc1114 1115 # Inside try so validation errors route through _handle_tool_errors1116 return self._normalize_tool_response(1117 response, request.tool_call, input_type1118 )1119 1120 # GraphInterrupt is a special exception that will always be raised.1121 # It can be triggered in the following scenarios,1122 # Where GraphInterrupt(GraphBubbleUp) is raised from an `interrupt` invocation1123 # most commonly:1124 # (1) a GraphInterrupt is raised inside a tool1125 # (2) a GraphInterrupt is raised inside a graph node for a graph called as a tool1126 # (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph1127 # called as a tool1128 # (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)1129 except GraphBubbleUp:1130 raise1131 except Exception as e:1132 # Determine which exception types are handled1133 handled_types: tuple[type[Exception], ...]1134 if isinstance(self._handle_tool_errors, type) and issubclass(1135 self._handle_tool_errors, Exception1136 ):1137 handled_types = (self._handle_tool_errors,)1138 elif isinstance(self._handle_tool_errors, tuple):1139 handled_types = self._handle_tool_errors1140 elif callable(self._handle_tool_errors) and not isinstance(1141 self._handle_tool_errors, type1142 ):1143 handled_types = _infer_handled_types(self._handle_tool_errors)1144 else:1145 # default behavior is catching all exceptions1146 handled_types = (Exception,)1147 1148 # Check if this error should be handled1149 if not self._handle_tool_errors or not isinstance(e, handled_types):1150 raise1151 1152 # Error is handled - create error ToolMessage1153 content = _handle_tool_error(e, flag=self._handle_tool_errors)1154 return ToolMessage(1155 content=content,1156 name=call["name"],1157 tool_call_id=call["id"],1158 status="error",1159 )1160 1161 async def _arun_one(1162 self,1163 call: ToolCall,1164 input_type: Literal["list", "dict", "tool_calls"],1165 tool_runtime: ToolRuntime,1166 ) -> ToolMessage | Command | list[Command | ToolMessage]:1167 """Execute single tool call asynchronously with awrap_tool_call wrapper if configured.1168 1169 Args:1170 call: Tool call dict.1171 input_type: Input format.1172 tool_runtime: Tool runtime.1173 1174 Returns:1175 ToolMessage or Command.1176 """1177 # Validation is deferred to _execute_tool_async to allow interceptors1178 # to short-circuit requests for unregistered tools1179 tool = self.tools_by_name.get(call["name"])1180 1181 # Create the tool request with state and runtime1182 tool_request = ToolCallRequest(1183 tool_call=call,1184 tool=tool,1185 state=tool_runtime.state,1186 runtime=tool_runtime,1187 )1188 1189 config = tool_runtime.config1190 1191 if self._awrap_tool_call is None and self._wrap_tool_call is None:1192 # No wrapper - execute directly1193 return await self._execute_tool_async(tool_request, input_type, config)1194 1195 # Define async execute callable that can be called multiple times1196 async def execute(req: ToolCallRequest) -> ToolMessage | Command:1197 """Execute tool with given request. Can be called multiple times."""1198 return await self._execute_tool_async(req, input_type, config)1199 1200 def _sync_execute(req: ToolCallRequest) -> ToolMessage | Command: