langgraph_agent_toolkit.agents.agent_executor module

async langgraph_agent_toolkit.agents.agent_executor.get_graph_history(graph, config)[source][source]

Read messages from StateGraph state or Functional API saved state.

Parameters:

  • graph (Pregel)

  • config (RunnableConfig)

Return type:

list[BaseMessage]

async langgraph_agent_toolkit.agents.agent_executor.add_graph_history(graph, config, messages)[source][source]

Append messages while preserving Functional API saved state.

Parameters:

  • graph (Pregel)

  • config (RunnableConfig)

  • messages (list[Any])

Return type:

None

langgraph_agent_toolkit.agents.agent_executor.interrupt_value_to_content(value)[source][source]

Convert an interrupt payload to valid AIMessage content.

Custom interrupt() blueprints pass a string. The function returns that string unchanged. HumanInTheLoopMiddleware passes a request dictionary. The function joins the action descriptions and adds reply instructions. This prevents an invalid dictionary value in AIMessage(content=...).

Parameters:

value (Any)

Return type:

Any

langgraph_agent_toolkit.agents.agent_executor.interrupts_to_chat_message(interrupts)[source][source]

Keep every interrupt ID and payload in the response.

Parameters:

interrupts (list[Interrupt])

Return type:

ChatMessage

langgraph_agent_toolkit.agents.agent_executor.build_resume_command(interrupted_tasks, user_input)[source][source]

Build the Command(resume=...) for an interrupted run.

HumanInTheLoopMiddleware expects {"decisions": [...]}. Other interrupt() blueprints read the input dictionary. For a HITL tool approval request, translate the user’s reply to a decision for each pending tool call. Otherwise, return the input dictionary unchanged.

Parameters:

  • interrupted_tasks (list)

  • user_input (Dict[str, Any])

Return type:

Command

class langgraph_agent_toolkit.agents.agent_executor.AgentExecutor(*args)[source][source]

Bases: object

Load, run, and save LangGraph agents.

Initialize the AgentExecutor and import agents.

Parameters:

*args – Import strings for the agents. Example: “langgraph_agent_toolkit.agents.blueprints.react.agent:react_agent”.

Raises:

ValueError – If no agents are provided.

__init__(*args)[source][source]

Initialize the AgentExecutor and import agents.

Parameters:

*args – Import strings for the agents. Example: “langgraph_agent_toolkit.agents.blueprints.react.agent:react_agent”.

Raises:

ValueError – If no agents are provided.

load_agents_from_imports(args)[source][source]

Import agents from the specified import strings.

Parameters:

args (tuple)

Return type:

None

get_agent(agent_id)[source][source]

Return the agent with the specified ID.

Parameters:

agent_id (str) – The ID of the agent.

Returns:

The requested Agent instance.

Raises:

KeyError – The agent ID is not found.

Return type:

Agent

get_all_agent_info()[source][source]

Return information about all available agents.

Returns:

AgentInfo objects with agent IDs and descriptions.

Return type:

list[AgentInfo]

add_agent(agent_id, agent)[source][source]

Add an agent to the executor.

Parameters:

  • agent_id (str) – The ID for the agent.

  • agent (Agent) – The Agent instance.

Return type:

None

static handle_agent_errors(func)[source][source]

Handle errors during agent execution.

Handle GraphRecursionError and other exceptions.

Parameters:

func (Callable[[...], T]) – The function to decorate.

Returns:

The decorated function.

Return type:

Callable[[…], T]

async invoke(agent_id, input, thread_id=None, user_id=None, model_name=None, model_provider=None, model_config_key=None, agent_config=None, recursion_limit=None)[source][source]

Run an agent with a message and return its response.

Parameters:

  • agent_id (str) – ID of the agent to run.

  • input (Dict[str, Any]) – User message for the agent.

  • thread_id (str | None) – Optional conversation thread ID.

  • user_id (str | None) – Optional user ID.

  • model_name (str | None) – Optional replacement model name.

  • model_provider (str | None) – Optional replacement model provider.

  • model_config_key (str | None) – Optional replacement model configuration key.

  • agent_config (Dict[str, Any] | None) – Optional agent configuration.

  • recursion_limit (int | None) – Optional limit for graph recursion.

Returns:

The agent response as a ChatMessage.

Return type:

ChatMessage

stream(agent_id, input, thread_id=None, user_id=None, model_name=None, model_provider=None, model_config_key=None, stream_tokens=True, agent_config=None, recursion_limit=None)[source][source]

Stream an agent response as tokens or messages.

Parameters:

  • agent_id (str) – ID of the agent to run.

  • input (Dict[str, Any]) – User message for the agent.

  • thread_id (str | None) – Optional conversation thread ID.

  • user_id (str | None) – Optional user ID.

  • model_name (str | None) – Optional replacement model name.

  • model_provider (str | None) – Optional replacement model provider.

  • model_config_key (str | None) – Optional replacement model configuration key.

  • stream_tokens (bool) – Stream individual tokens when true.

  • agent_config (Dict[str, Any] | None) – Optional agent configuration.

  • recursion_limit (int | None) – Optional limit for graph recursion.

Yields:

Full ChatMessage objects or token strings.

Return type:

AsyncGenerator[str | ChatMessage, None]

save(path, agent_ids=None)[source][source]

Save agents to disk with joblib.

Parameters:

  • path (str) – Directory path for the agent files.

  • agent_ids (List[str] | None) – Agent IDs to save. Save all agents when this is None.

Return type:

None

load_saved_agents(path)[source][source]

Load agents from joblib files on disk.

Parameters:

path (str) – Directory path for the agent files.

Return type:

None