Mcpagent

View the source code for this module on GitHub: https://github.com/mcp-use/mcp-use/blob/main/libraries/python/mcp_use/agents/mcpagent.py

MCP: Main integration module with customizable system prompt. This module provides the main MCPAgent class that integrates all components to provide a simple interface for using MCP tools with different LLMs. LangChain 1.0.0 Migration:

The agent uses create_agent() from langchain.agents which returns a CompiledStateGraph
New methods: astream_simplified() and run_v2() leverage the built-in astream() from CompiledStateGraph which handles the agent loop internally
Legacy methods: stream() and run() use manual step-by-step execution for backward compatibility

from mcp_use.agents.mcpagent import MCPAgent

`method` init

Initialize a new MCPAgent instance.Parameters

langchain_core.language_models.base.BaseLanguageModel | None
default:"None"
The LangChain LLM to use. Not required if agent_id is provided for remote execution.

mcp_use.client.client.MCPClient | None
default:"None"
The MCPClient to use. If provided, connector is ignored.

list[mcp_use.client.connectors.base.BaseConnector] | None
default:"None"
A list of MCP connectors to use if client is not provided.

int
default:"5"
The maximum number of steps to take.

bool
default:"False"
Whether to automatically initialize the agent when run is called.

bool
default:"True"
Whether to maintain conversation history for context.

str | None
default:"None"
Complete system prompt to use (overrides template if provided).

str | None
default:"None"
Template for system prompt with placeholder.

str | None
default:"None"
Extra instructions to append to the system prompt.

list[str] | None
default:"None"
List of tool names that should not be available to the agent.

list[str] | None
default:"None"
List of tools

bool
default:"False"
Whether to use server manager mode instead of exposing all tools.

mcp_use.agents.managers.base.BaseServerManager | None
default:"None"
Server name or configuration

bool
default:"False"
Enable debug/verbose mode

bool
default:"False"
Whether to pretty print the output.

str | None
default:"None"
Remote agent ID for remote execution. If provided, creates a remote agent.

str | None
default:"None"
API key for remote execution. If None, checks MCP_USE_API_KEY env var.

str
default:"https://cloud.manufact.com"
Base URL for remote API calls.

list | None
default:"None"
List of LangChain callbacks to use. If None and Langfuse is configured, uses langfuse_handler.

str | None
default:"None"
String value

bool
default:"True"
Whether to enable automatic error handling for tool calls. When True, tool errors

Signature

def __init__(llm: langchain_core.language_models.base.BaseLanguageModel | None = None, client: mcp_use.client.client.MCPClient | None = None, connectors: list[mcp_use.client.connectors.base.BaseConnector] | None = None, max_steps: int = 5, auto_initialize: bool = False, memory_enabled: bool = True, system_prompt: str | None = None, system_prompt_template: str | None = None, additional_instructions: str | None = None, disallowed_tools: list[str] | None = None, tools_used_names: list[str] | None = None, use_server_manager: bool = False, server_manager: mcp_use.agents.managers.base.BaseServerManager | None = None, verbose: bool = False, pretty_print: bool = False, agent_id: str | None = None, api_key: str | None = None, base_url: str = "https://cloud.manufact.com", callbacks: list | None = None, chat_id: str | None = None, retry_on_error: bool = True):

`method` add_to_history

Add a message to the conversation history.Parameters

langchain_core.messages.base.BaseMessage
required
The message to add.

Signature

def add_to_history(message: langchain_core.messages.base.BaseMessage):

`method` clear_conversation_history

Clear the conversation history.Signature

def clear_conversation_history():

`method` close

Close the MCP connection with improved error handling.Signature

def close():

`method` get_conversation_history

Get the current conversation history.Returns

list[langchain_core.messages.base.BaseMessage]
The list of conversation messages.

Signature

def get_conversation_history():

`method` get_disallowed_tools

Get the list of tools that are not available to the agent.Returns

list[str]
List of tool names that are not available.

Signature

def get_disallowed_tools():

`method` get_system_message

Get the current system message.Returns

langchain_core.messages.system.SystemMessage | None
The current system message, or None if not set.

Signature

def get_system_message():

`method` initialize

Initialize the MCP client and agent.Signature

def initialize():

`method` run

Run a query using LangChain 1.0.0’s agent and return the final result.Example:

# Regular usage
result = await agent.run("What's the weather like?")

# Structured output usage
from pydantic import BaseModel, Field

class WeatherInfo(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    condition: str = Field(description="Weather condition")

weather: WeatherInfo = await agent.run(
    "What's the weather like?",
    output_schema=WeatherInfo
)

Parameters

str | langchain_core.messages.human.HumanMessage
required
The query to run. Accepts a plain string or a HumanMessage when

int | None
default:"None"
Optional maximum number of steps to take.

bool
default:"True"
Whether to handle the connector lifecycle internally.

list[langchain_core.messages.base.BaseMessage] | None
default:"None"
Optional external history to use instead of the

type[~T] | None
default:"None"
Optional Pydantic BaseModel class for structured output.

Returns

str | mcp_use.agents.mcpagent.T
The result of running the query as a string, or if output_schema is provided, an instance of the specified Pydantic model.

Signature

def run(
query: str | langchain_core.messages.human.HumanMessage,
    max_steps: int | None = None,
    manage_connector: bool = True,
    external_history: list[langchain_core.messages.base.BaseMessage] | None = None,
    output_schema: type[~T] | None = None
):

`method` set_disallowed_tools

Set the list of tools that should not be available to the agent.This will take effect the next time the agent is initialized.Parameters

list[str]
required
List of tool names that should not be available.

Signature

def set_disallowed_tools(disallowed_tools: list[str]):

`method` set_system_message

Set a new system message.Parameters

str
required
The new system message content.

Signature

def set_system_message(message: str):

`method` stream

Async generator using LangChain 1.0.0’s create_agent and astream.This method leverages the LangChain 1.0.0 API where create_agent returns a CompiledStateGraph that handles the agent loop internally via astream.Tool Updates with Server Manager: When using server_manager mode, this method handles dynamic tool updates:

Before execution: Updates are applied immediately to the new stream
During execution: When tools change, we wait for a “safe restart point” (after tool results complete), then interrupt the stream, recreate the agent with new tools, and resume execution with accumulated messages.
Safe restart points: Only restart after tool results to ensure message pairs (tool_use + tool_result) are complete, satisfying LLM API requirements.
Max restarts: Limited to 3 restarts to prevent infinite loops

This interrupt-and-restart approach ensures that tools added mid-execution (e.g., via connect_to_mcp_server) are immediately available to the agent, maintaining the same behavior as the legacy implementation while respecting API constraints.Yields: Intermediate steps and final result from the agent execution.Parameters

str | langchain_core.messages.human.HumanMessage
required
The query to run. Accepts a plain string or a HumanMessage when

int | None
default:"None"
Integer value

bool
default:"True"
Whether to handle the connector lifecycle internally.

list[langchain_core.messages.base.BaseMessage] | None
default:"None"
Optional external history to use instead of the

bool
default:"True"
Boolean flag

type[~T] | None
default:"None"
Optional Pydantic BaseModel class for structured output.

Returns

AsyncGenerator

Signature

def stream(
query: str | langchain_core.messages.human.HumanMessage,
    max_steps: int | None = None,
    manage_connector: bool = True,
    external_history: list[langchain_core.messages.base.BaseMessage] | None = None,
    track_execution: bool = True,
    output_schema: type[~T] | None = None
):

`method` stream_events

Asynchronous streaming interface.Example::async for chunk in agent.stream_events(“hello”): print(chunk)Parameters

str | langchain_core.messages.human.HumanMessage
required
Query string or input

int | None
default:"None"
Integer value

bool
default:"True"
Connector instance

list[langchain_core.messages.base.BaseMessage] | None
default:"None"
List of items

Returns

AsyncIterator

Signature

def stream_events(
query: str | langchain_core.messages.human.HumanMessage,
    max_steps: int | None = None,
    manage_connector: bool = True,
    external_history: list[langchain_core.messages.base.BaseMessage] | None = None
):

Agents

Client

Core

Errors

Server

Mcpagent

MCPAgent

`method` init

`method` add_to_history

`method` clear_conversation_history

`method` close

`method` get_conversation_history

`method` get_disallowed_tools

`method` get_system_message

`method` initialize

`method` run

`method` set_disallowed_tools

`method` set_system_message

`method` stream

`method` stream_events

​MCPAgent

​method init

​method add_to_history

​method clear_conversation_history

​method close

​method get_conversation_history

​method get_disallowed_tools

​method get_system_message

​method initialize

​method run

​method set_disallowed_tools

​method set_system_message

​method stream

​method stream_events

MCPAgent

`method` init

`method` add_to_history

`method` clear_conversation_history

`method` close

`method` get_conversation_history

`method` get_disallowed_tools

`method` get_system_message

`method` initialize

`method` run

`method` set_disallowed_tools

`method` set_system_message

`method` stream

`method` stream_events