Base langgraph agent

Base class for LangGraph-based agent implementations.

This class provides the core LangGraph machinery including graph compilation, state handling, and I/O mapping for LangGraph agents.

Authors

Christian Trisno Sen Long Chen (christian.t.s.l.chen@gdplabs.id) Putu Ravindra Wiguna (putu.r.wiguna@gdplabs.id)

BaseLangGraphAgent(name, instruction, description=None, model=None, tools=None, state_schema=None, thread_id_key='thread_id', event_emitter=None, checkpointer=None, **kwargs)

Bases: BaseAgent

Base class for LangGraph-based agents with unified tool approach.

Provides core LangGraph functionality including: - Graph compilation and execution - State schema management - I/O mapping between user inputs and graph states - Event emission support - Tool resolution and handling - A2A communication capabilities via tools - Agent delegation capabilities via tools - MCP server integration via tools - Enhanced output extraction from various state formats

Tool Management: - regular_tools: Standard LangChain tools provided during initialization - mcp_tools: Tools retrieved from MCP servers - resolved_tools: Combined collection of all tools for graph execution

Subclasses must implement: - define_graph(): Define the specific graph structure - _prepare_graph_input(): Convert user input to graph state - _format_graph_output(): Convert final graph state to user output

Initialize the BaseLangGraphAgent.

Parameters:

Name	Type	Description	Default
name	str	The name of the agent.	required
instruction	str	The system instruction for the agent.	required
description	str | None	Human-readable description of the agent.	None
model	Any | None	The model to use (lm_invoker, LangChain model, string, etc.).	None
tools	Sequence[BaseTool] | None	Sequence of regular LangChain tools (not A2A or delegation tools).	None
state_schema	Type | None	The state schema for the LangGraph. Defaults to basic message state.	None
thread_id_key	str	Key for thread ID in configuration.	'thread_id'
event_emitter	EventEmitter | None	Optional event emitter for streaming updates.	None
checkpointer	Checkpointer | None	Optional checkpointer for conversation persistence.	None
**kwargs	Any	Additional keyword arguments passed to BaseAgent.	{}

add_mcp_server(mcp_config)

Add MCP server configuration.

Parameters:

Name	Type	Description	Default
mcp_config	dict[str, dict[str, Any]]	Dictionary containing MCP server configurations.	required

arun(query, **kwargs) async

Asynchronously run the LangGraph agent.

If MCP configuration exists, connects to the MCP server and registers tools before running.

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
**kwargs	Any	Additional keyword arguments including configurable for LangGraph.	{}

Returns:

Type	Description
dict[str, Any]	Dictionary containing the agent's response and full final state.

arun_a2a_stream(query, **kwargs) async

Asynchronously streams the agent's response in a generic format for A2A.

If MCP configuration exists, connects to the MCP server and registers tools before streaming.

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Type	Description
AsyncGenerator[dict[str, Any], None]	Dictionaries with "status" and "content" keys.
AsyncGenerator[dict[str, Any], None]	Possible statuses: "working", "completed", "failed", "canceled".

arun_stream(query, **kwargs) async

Asynchronously stream the LangGraph agent's response.

If MCP configuration exists, connects to the MCP server and registers tools before streaming. This method properly handles both LM Invoker and LangChain model streaming: - For LM Invoker: Uses StreamEventHandler to capture streaming events - For LangChain models: Uses LangGraph's native streaming implementation

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Type	Description
AsyncGenerator[str | dict[str, Any], None]	Chunks of output (strings or dicts) from the streaming response.

define_graph(graph_builder) abstractmethod

Define the specific graph structure for this agent type.

Subclasses must implement this method to: 1. Add nodes to the graph_builder 2. Add edges and conditional edges 3. Set entry points 4. Return the compiled graph

Parameters:

Name	Type	Description	Default
graph_builder	StateGraph	The StateGraph builder to define nodes and edges on.	required

Returns:

Type	Description
CompiledStateGraph	The compiled graph ready for execution.

register_a2a_agents(agent_cards)

Register A2A communication capabilities using the A2A tool manager.

Parameters:

Name	Type	Description	Default
agent_cards	list[AgentCard]	List of AgentCard instances for external communication.	required

register_delegation_agents(agents)

Register internal agent delegation capabilities using the delegation tool manager.

Parameters:

Name	Type	Description	Default
agents	list[BaseAgent]	List of BaseAgent instances for internal task delegation.	required

run(query, **kwargs)

Synchronously run the LangGraph agent.

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
dict[str, Any]	Dictionary containing the agent's response.

update_regular_tools(new_tools, rebuild_graph=None)

Update regular tools (not capabilities).

Parameters:

Name	Type	Description	Default
new_tools	list[BaseTool]	New list of regular tools to use.	required
rebuild_graph	bool | None	Whether to rebuild graph. If None, uses auto_rebuild_graph setting.	None