jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

fix: add error handling and message on mcp servers, fixed tool mode on mcp (#8717)

Browse source 
* Added mcp error handling

* Added error param to servers

* Added error display on mcp servers page

* Added error display on mcp component

* Added error handling for other types of errors

* uv lock update and add error handling

* update mcp version

* fixed tool mode not working on mcp component

* update mcp to 1.9.4

* Removed server parameters custom commands on connection to stdio

* removed unused import

* disable mcp notice

* Removed drop state when home type is mcp

* Added loading before showing tools

* Updated mcp to 1.9.4

* Decreased mcp timeout

* Implemented error surfacing with exec command

* removed non default keys either way when mcp_server is used

* update to session handling

* [autofix.ci] apply automated fixes

* updated it to check if it is the same server to not clear on startup

* update to components

* Update mcp_component.py

* Update mcp_component.py

* Update mcp_component.py

---------

Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
This commit is contained in:
Lucas Oliveira 2025-06-26 19:22:01 -03:00 • committed by GitHub
commit f4d761e63d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 178 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

45
src/backend/base/langflow/api/v2/mcp.py
View file

@ -122,13 +122,52 @@ async def get_servers(
)
server_info["mode"] = mode.lower()
server_info["toolsCount"] = len(tool_list)
if len(tool_list) == 0:
server_info["error"] = "No tools found"
except ValueError as e:
# Configuration validation errors, invalid URLs, etc.
logger.error(f"Configuration error for server {server_name}: {e}")
server_info["error"] = f"Configuration error: {e}"
except (ConnectionError, TimeoutError) as e:
# Network connection and timeout issues
logger.error(f"Connection error for server {server_name}: {e}")
server_info["error"] = f"Connection failed: {e}"
except OSError as e:
# System-level errors (process execution, file access)
logger.error(f"System error for server {server_name}: {e}")
server_info["error"] = f"System error: {e}"
except (KeyError, TypeError) as e:
# Data parsing and access errors
logger.error(f"Data error for server {server_name}: {e}")
server_info["error"] = f"Configuration data error: {e}"
except (RuntimeError, ProcessLookupError, PermissionError) as e:
# Runtime and process-related errors
logger.error(f"Runtime error for server {server_name}: {e}")
server_info["error"] = f"Runtime error: {e}"
except Exception as e: # noqa: BLE001
logger.exception(f"Error checking server {server_name}: {e}")
# Generic catch-all for other exceptions (including ExceptionGroup)
if hasattr(e, "exceptions") and e.exceptions:
# Extract the first underlying exception for a more meaningful error message
underlying_error = e.exceptions[0]
logger.exception(f"Error checking server {server_name}: {underlying_error}")
server_info["error"] = f"Error loading server: {underlying_error}"
else:
logger.exception(f"Error checking server {server_name}: {e}")
server_info["error"] = f"Error loading server: {e}"
return server_info
async def check_server_with_timeout(server_name: str) -> dict:
try:
return await asyncio.wait_for(
check_server(server_name), timeout=get_settings_service().settings.mcp_server_timeout
)
except asyncio.TimeoutError:
logger.error(f"Timeout checking server {server_name}")
return {"name": server_name, "mode": None, "toolsCount": None, "error": "Server check timed out."}
# Run all server checks concurrently
tasks = [check_server(server) for server in server_list["mcpServers"]]
return await asyncio.gather(*tasks, return_exceptions=False)
tasks = [check_server_with_timeout(server) for server in server_list["mcpServers"]]
return await asyncio.gather(*tasks, return_exceptions=True)
@router.get("/servers/{server_name}")

43
src/backend/base/langflow/base/mcp/util.py
View file

@ -16,6 +16,7 @@ from pydantic import BaseModel, Field, create_model
from sqlmodel import select
from langflow.services.database.models.flow.model import Flow
from langflow.services.deps import get_settings_service
HTTP_ERROR_STATUS_CODE = httpx_codes.BAD_REQUEST # HTTP status code for client errors
NULLABLE_TYPE_LENGTH = 2 # Number of types in a nullable union (the type itself + null)
@ -300,7 +301,7 @@ class MCPStdioClient:
self._connection_params = None
self._connected = False
async def connect_to_server(self, command_str: str, env: dict[str, str] | None = None) -> list[StructuredTool]:
async def _connect_to_server(self, command_str: str, env: dict[str, str] | None = None) -> list[StructuredTool]:
"""Connect to MCP server using stdio transport (SDK style)."""
from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
@ -320,7 +321,7 @@ class MCPStdioClient:
else:
server_params = StdioServerParameters(
command="bash",
args=["-c", f"{command_str} || echo 'Command failed with exit code $?' >&2"],
args=["-c", f"exec {command_str} || echo 'Command failed with exit code $?' >&2"],
env=env_data,
)
@ -339,6 +340,19 @@ class MCPStdioClient:
self._connected = False
return []
async def connect_to_server(self, command_str: str, env: dict[str, str] | None = None) -> list[StructuredTool]:
"""Connect to MCP server using stdio transport (SDK style)."""
try:
return await asyncio.wait_for(
self._connect_to_server(command_str, env), timeout=get_settings_service().settings.mcp_server_timeout
)
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
logger.error(f"Failed to connect to MCP stdio server: {e}")
self._connection_params = None
self._connected = False
msg = f"Failed to connect to MCP stdio server: {e}"
raise ValueError(msg) from e
async def disconnect(self):
"""Properly close the connection and clean up resources."""
self.session = None
@ -425,7 +439,7 @@ class MCPSseClient:
logger.warning(f"Error checking redirects: {e}")
return url
async def connect_to_server(
async def _connect_to_server(
self,
url: str | None,
headers: dict[str, str] | None = None,
@ -464,6 +478,19 @@ class MCPSseClient:
response = await session.list_tools()
self._connected = True
return response.tools
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
logger.error(f"Failed to connect to MCP SSE server: {e}")
self._connection_params = None
self._connected = False
msg = f"Failed to connect to MCP SSE server: {e}"
raise ValueError(msg) from e
async def connect_to_server(self, url: str, headers: dict[str, str] | None = None) -> list[StructuredTool]:
"""Connect to MCP server using SSE transport (SDK style)."""
try:
return await asyncio.wait_for(
self._connect_to_server(url, headers), timeout=get_settings_service().settings.mcp_server_timeout
)
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
logger.error(f"Failed to connect to MCP SSE server: {e}")
self._connection_params = None
@ -568,7 +595,10 @@ async def update_tools(
return "", [], {}
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
logger.error(f"Failed to connect to MCP server '{server_name}': {e}")
return "", [], {}
# return "", [], {}
msg = f"Failed to connect to MCP server '{server_name}': {e}"
logger.error(msg)
raise ValueError(msg) from e
if not tools or not client or not client._connected:
logger.warning(f"No tools available from MCP server '{server_name}' or connection failed")
@ -598,10 +628,11 @@ async def update_tools(
tool_cache[tool.name] = tool_obj
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
logger.error(f"Failed to create tool '{tool.name}' from server '{server_name}': {e}")
continue
msg = f"Failed to create tool '{tool.name}' from server '{server_name}': {e}"
raise ValueError(msg) from e
logger.info(f"Successfully loaded {len(tool_list)} tools from MCP server '{server_name}'")
except (ConnectionError, TimeoutError, OSError, ValueError) as e:
except (ConnectionError, TimeoutError, OSError, ValueError, AttributeError, AssertionError) as e:
logger.error(f"Unexpected error while updating tools for MCP server '{server_name}': {e}")
return "", [], {}
else:

79
src/backend/base/langflow/components/data/mcp_component.py
View file

@ -67,6 +67,7 @@ class MCPToolsComponent(ComponentWithCache):
sse_client: MCPSseClient = MCPSseClient()
tools: list = []
_tool_cache: dict = {}
_last_selected_server: str | None = None # Cache for the last selected server
default_keys: list[str] = [
"code",
"_type",
@ -217,12 +218,14 @@ class MCPToolsComponent(ComponentWithCache):
if len(self.tools) == 0:
try:
self.tools, build_config["mcp_server"]["value"] = await self.update_tool_list()
build_config["tool"]["options"] = [tool.name for tool in self.tools]
except ValueError:
build_config["tool"]["options"] = []
build_config["tool"]["value"] = ""
build_config["tool"]["placeholder"] = "Error on MCP Server"
return build_config
build_config["tool"]["placeholder"] = ""
if field_value == "":
return build_config
tool_obj = None
@ -242,30 +245,61 @@ class MCPToolsComponent(ComponentWithCache):
else:
return build_config
elif field_name == "mcp_server":
try:
# field_value is now a dict {name, config}
mcp_server_value = field_value
self.tools, build_config["mcp_server"]["value"] = await self.update_tool_list(mcp_server_value)
except ValueError:
if not build_config["tools_metadata"]["show"]:
build_config["tool"]["show"] = True
build_config["tool"]["options"] = []
build_config["tool"]["value"] = ""
build_config["tool"]["placeholder"] = "Error on MCP Server"
else:
build_config["tool"]["show"] = False
self.remove_non_default_keys(build_config)
return build_config
build_config["tool"]["placeholder"] = ""
if "tool" in build_config and len(self.tools) > 0 and not build_config["tools_metadata"]["show"]:
build_config["tool"]["show"] = True
build_config["tool"]["options"] = [tool.name for tool in self.tools]
await self._update_tool_config(build_config, build_config["tool"]["value"])
elif "tool" in build_config and len(self.tools) == 0:
self.remove_non_default_keys(build_config)
if not field_value:
build_config["tool"]["show"] = False
build_config["tool"]["options"] = []
build_config["tool"]["value"] = ""
build_config["tool"]["placeholder"] = ""
self.remove_non_default_keys(build_config)
return build_config
current_server_name = field_value.get("name") if isinstance(field_value, dict) else field_value
# To avoid unnecessary updates, only proceed if the server has actually changed
if self._last_selected_server == current_server_name:
return build_config
# Determine if "Tool Mode" is active by checking if the tool dropdown is hidden.
# The check for _last_selected_server handles the initial state where the dropdown is
# hidden by default but we are not yet in "Tool Mode".
is_in_tool_mode = build_config["tools_metadata"]["value"]
self._last_selected_server = current_server_name
self.tools = [] # Clear previous tools
self.remove_non_default_keys(build_config) # Clear previous tool inputs
# Only show the tool dropdown if not in tool_mode
if not is_in_tool_mode:
build_config["tool"]["show"] = True
build_config["tool"]["placeholder"] = "Loading tools..."
build_config["tool"]["options"] = []
build_config["tool"]["value"] = ""
else:
# Keep the tool dropdown hidden if in tool_mode
build_config["tool"]["show"] = False
try:
# Fetch tools for the newly selected server
tools, server_info = await self.update_tool_list(field_value)
build_config["mcp_server"]["value"] = server_info
if tools:
tool_names = [tool.name for tool in tools]
if not is_in_tool_mode:
build_config["tool"]["options"] = tool_names
build_config["tool"]["placeholder"] = "Select a tool"
elif not is_in_tool_mode:
build_config["tool"]["placeholder"] = "No tools found"
except (ValueError, AttributeError, KeyError, TypeError, ConnectionError, TimeoutError) as e:
logger.error(f"Failed to fetch tools for server '{current_server_name}': {e}")
if not is_in_tool_mode:
build_config["tool"]["placeholder"] = "Error fetching tools"
build_config["tool"]["options"] = []
build_config["tool"]["value"] = ""
finally:
# Ensure we don't show inputs for a tool that might no longer be valid
await self._update_tool_config(build_config, "")
elif field_name == "tool_mode":
try:
self.tools, build_config["mcp_server"]["value"] = await self.update_tool_list()
@ -429,4 +463,5 @@ class MCPToolsComponent(ComponentWithCache):
async def _get_tools(self):
"""Get cached tools or update if necessary."""
mcp_server = getattr(self, "mcp_server", None)
return await self.update_tool_list(mcp_server)
tools, _ = await self.update_tool_list(mcp_server)
return tools

4
src/backend/base/langflow/services/settings/base.py
View file

@ -92,6 +92,10 @@ class Settings(BaseSettings):
"""The number of seconds to wait before giving up on a lock to released or establishing a connection to the
database."""
mcp_server_timeout: int = 20
"""The number of seconds to wait before giving up on a lock to released or establishing a connection to the
database."""
# sqlite configuration
sqlite_pragmas: dict | None = {"synchronous": "NORMAL", "journal_mode": "WAL"}
"""SQLite pragmas to use when connecting to the database."""

2
src/backend/base/pyproject.toml
View file

@ -77,7 +77,7 @@ dependencies = [
"validators>=0.34.0",
"networkx>=3.4.2",
"json-repair>=0.30.3",
"mcp>=1.6.0",
"mcp~=1.9.4",
"aiosqlite>=0.20.0",
"greenlet>=3.1.1",
"jsonquerylang>=1.1.1",