diff --git a/Dockerfile.langflow b/Dockerfile.langflow index 71baf447..bdae1f70 100644 --- a/Dockerfile.langflow +++ b/Dockerfile.langflow @@ -1,4 +1,4 @@ -FROM langflowai/langflow-nightly:1.6.3.dev0 +FROM langflowai/langflow-nightly:1.6.3.dev1 EXPOSE 7860 diff --git a/docker-compose-cpu.yml b/docker-compose-cpu.yml index 0c09254a..40507c94 100644 --- a/docker-compose-cpu.yml +++ b/docker-compose-cpu.yml @@ -108,6 +108,7 @@ services: - OWNER_NAME=None - OWNER_EMAIL=None - CONNECTOR_TYPE=system + - CONNECTOR_TYPE_URL=url - OPENRAG-QUERY-FILTER="{}" - OPENSEARCH_PASSWORD=${OPENSEARCH_PASSWORD} - FILENAME=None diff --git a/docker-compose.yml b/docker-compose.yml index be9bcbc9..4a68d210 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -81,7 +81,6 @@ services: # build: # context: . # dockerfile: Dockerfile.frontend - #dockerfile: Dockerfile.frontend container_name: openrag-frontend depends_on: - openrag-backend @@ -109,6 +108,7 @@ services: - OWNER_NAME=None - OWNER_EMAIL=None - CONNECTOR_TYPE=system + - CONNECTOR_TYPE_URL=url - OPENRAG-QUERY-FILTER="{}" - FILENAME=None - MIMETYPE=None diff --git a/flows/openrag_agent.json b/flows/openrag_agent.json index c08a305c..bb02b425 100644 --- a/flows/openrag_agent.json +++ b/flows/openrag_agent.json @@ -170,6 +170,31 @@ "sourceHandle": "{œdataTypeœ:œTextInputœ,œidœ:œTextInput-aHsQbœ,œnameœ:œtextœ,œoutput_typesœ:[œMessageœ]}", "target": "OpenSearch-iYfjf", "targetHandle": "{œfieldNameœ:œfilter_expressionœ,œidœ:œOpenSearch-iYfjfœ,œinputTypesœ:[œMessageœ],œtypeœ:œstrœ}" + }, + { + "data": { + "sourceHandle": { + "dataType": "MCP", + "id": "MCP-7EY21", + "name": "component_as_tool", + "output_types": [ + "Tool" + ] + }, + "targetHandle": { + "fieldName": "tools", + "id": "Agent-crjWf", + "inputTypes": [ + "Tool" + ], + "type": "other" + } + }, + "id": "xy-edge__MCP-7EY21{œdataTypeœ:œMCPœ,œidœ:œMCP-7EY21œ,œnameœ:œcomponent_as_toolœ,œoutput_typesœ:[œToolœ]}-Agent-crjWf{œfieldNameœ:œtoolsœ,œidœ:œAgent-crjWfœ,œinputTypesœ:[œToolœ],œtypeœ:œotherœ}", + "source": "MCP-7EY21", + "sourceHandle": "{œdataTypeœ:œMCPœ,œidœ:œMCP-7EY21œ,œnameœ:œcomponent_as_toolœ,œoutput_typesœ:[œToolœ]}", + "target": "Agent-crjWf", + "targetHandle": "{œfieldNameœ:œtoolsœ,œidœ:œAgent-crjWfœ,œinputTypesœ:[œToolœ],œtypeœ:œotherœ}" } ], "nodes": [ @@ -730,7 +755,7 @@ ], "frozen": false, "icon": "OpenSearch", - "last_updated": "2025-10-04T05:41:33.344Z", + "last_updated": "2025-10-06T15:23:50.339Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -1384,7 +1409,7 @@ ], "frozen": false, "icon": "binary", - "last_updated": "2025-10-04T05:41:33.345Z", + "last_updated": "2025-10-06T15:23:50.341Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -1709,7 +1734,7 @@ ], "frozen": false, "icon": "bot", - "last_updated": "2025-10-04T05:41:33.399Z", + "last_updated": "2025-10-06T15:23:50.396Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -2248,7 +2273,7 @@ ], "frozen": false, "icon": "brain-circuit", - "last_updated": "2025-10-04T05:41:33.347Z", + "last_updated": "2025-10-06T15:23:50.343Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -2551,10 +2576,258 @@ }, "selected": false, "type": "genericNode" + }, + { + "data": { + "id": "MCP-7EY21", + "node": { + "base_classes": [ + "DataFrame" + ], + "beta": false, + "category": "MCP", + "conditional_paths": [], + "custom_fields": {}, + "description": "Connect to an MCP server to use its tools.", + "display_name": "MCP Tools", + "documentation": "https://docs.langflow.org/mcp-client", + "edited": false, + "field_order": [ + "mcp_server", + "use_cache", + "tool", + "tool_placeholder" + ], + "frozen": false, + "icon": "Mcp", + "key": "mcp_lf-starter_project", + "last_updated": "2025-10-06T15:23:56.578Z", + "legacy": false, + "mcpServerName": "lf-starter_project", + "metadata": { + "code_hash": "756d1e10d0ca", + "dependencies": { + "dependencies": [ + { + "name": "langchain_core", + "version": "0.3.77" + }, + { + "name": "lfx", + "version": null + }, + { + "name": "langflow", + "version": null + } + ], + "total_dependencies": 3 + }, + "module": "lfx.components.agents.mcp_component.MCPToolsComponent" + }, + "minimized": false, + "output_types": [], + "outputs": [ + { + "allows_loop": false, + "cache": true, + "display_name": "Toolset", + "group_outputs": false, + "hidden": null, + "method": "to_toolkit", + "name": "component_as_tool", + "options": null, + "required_inputs": null, + "selected": "Tool", + "tool_mode": true, + "types": [ + "Tool" + ], + "value": "__UNDEFINED__" + } + ], + "pinned": false, + "template": { + "_type": "Component", + "code": { + "advanced": true, + "dynamic": true, + "fileTypes": [], + "file_path": "", + "info": "", + "list": false, + "load_from_db": false, + "multiline": true, + "name": "code", + "password": false, + "placeholder": "", + "required": true, + "show": true, + "title_case": false, + "type": "code", + "value": "from __future__ import annotations\n\nimport asyncio\nimport uuid\nfrom typing import Any\n\nfrom langchain_core.tools import StructuredTool # noqa: TC002\n\nfrom lfx.base.agents.utils import maybe_unflatten_dict, safe_cache_get, safe_cache_set\nfrom lfx.base.mcp.util import MCPSseClient, MCPStdioClient, create_input_schema_from_json_schema, update_tools\nfrom lfx.custom.custom_component.component_with_cache import ComponentWithCache\nfrom lfx.inputs.inputs import InputTypes # noqa: TC001\nfrom lfx.io import BoolInput, DropdownInput, McpInput, MessageTextInput, Output\nfrom lfx.io.schema import flatten_schema, schema_to_langflow_inputs\nfrom lfx.log.logger import logger\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.services.deps import get_settings_service, get_storage_service, session_scope\n\n\nclass MCPToolsComponent(ComponentWithCache):\n schema_inputs: list = []\n tools: list[StructuredTool] = []\n _not_load_actions: bool = False\n _tool_cache: dict = {}\n _last_selected_server: str | None = None # Cache for the last selected server\n\n def __init__(self, **data) -> None:\n super().__init__(**data)\n # Initialize cache keys to avoid CacheMiss when accessing them\n self._ensure_cache_structure()\n\n # Initialize clients with access to the component cache\n self.stdio_client: MCPStdioClient = MCPStdioClient(component_cache=self._shared_component_cache)\n self.sse_client: MCPSseClient = MCPSseClient(component_cache=self._shared_component_cache)\n\n def _ensure_cache_structure(self):\n \"\"\"Ensure the cache has the required structure.\"\"\"\n # Check if servers key exists and is not CacheMiss\n servers_value = safe_cache_get(self._shared_component_cache, \"servers\")\n if servers_value is None:\n safe_cache_set(self._shared_component_cache, \"servers\", {})\n\n # Check if last_selected_server key exists and is not CacheMiss\n last_server_value = safe_cache_get(self._shared_component_cache, \"last_selected_server\")\n if last_server_value is None:\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", \"\")\n\n default_keys: list[str] = [\n \"code\",\n \"_type\",\n \"tool_mode\",\n \"tool_placeholder\",\n \"mcp_server\",\n \"tool\",\n \"use_cache\",\n ]\n\n display_name = \"MCP Tools\"\n description = \"Connect to an MCP server to use its tools.\"\n documentation: str = \"https://docs.langflow.org/mcp-client\"\n icon = \"Mcp\"\n name = \"MCPTools\"\n\n inputs = [\n McpInput(\n name=\"mcp_server\",\n display_name=\"MCP Server\",\n info=\"Select the MCP Server that will be used by this component\",\n real_time_refresh=True,\n ),\n BoolInput(\n name=\"use_cache\",\n display_name=\"Use Cached Server\",\n info=(\n \"Enable caching of MCP Server and tools to improve performance. \"\n \"Disable to always fetch fresh tools and server updates.\"\n ),\n value=False,\n advanced=True,\n ),\n DropdownInput(\n name=\"tool\",\n display_name=\"Tool\",\n options=[],\n value=\"\",\n info=\"Select the tool to execute\",\n show=False,\n required=True,\n real_time_refresh=True,\n ),\n MessageTextInput(\n name=\"tool_placeholder\",\n display_name=\"Tool Placeholder\",\n info=\"Placeholder for the tool\",\n value=\"\",\n show=False,\n tool_mode=False,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Response\", name=\"response\", method=\"build_output\"),\n ]\n\n async def _validate_schema_inputs(self, tool_obj) -> list[InputTypes]:\n \"\"\"Validate and process schema inputs for a tool.\"\"\"\n try:\n if not tool_obj or not hasattr(tool_obj, \"args_schema\"):\n msg = \"Invalid tool object or missing input schema\"\n raise ValueError(msg)\n\n flat_schema = flatten_schema(tool_obj.args_schema.schema())\n input_schema = create_input_schema_from_json_schema(flat_schema)\n if not input_schema:\n msg = f\"Empty input schema for tool '{tool_obj.name}'\"\n raise ValueError(msg)\n\n schema_inputs = schema_to_langflow_inputs(input_schema)\n if not schema_inputs:\n msg = f\"No input parameters defined for tool '{tool_obj.name}'\"\n await logger.awarning(msg)\n return []\n\n except Exception as e:\n msg = f\"Error validating schema inputs: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return schema_inputs\n\n async def update_tool_list(self, mcp_server_value=None):\n # Accepts mcp_server_value as dict {name, config} or uses self.mcp_server\n mcp_server = mcp_server_value if mcp_server_value is not None else getattr(self, \"mcp_server\", None)\n server_name = None\n server_config_from_value = None\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\")\n server_config_from_value = mcp_server.get(\"config\")\n else:\n server_name = mcp_server\n if not server_name:\n self.tools = []\n return [], {\"name\": server_name, \"config\": server_config_from_value}\n\n # Check if caching is enabled, default to False\n use_cache = getattr(self, \"use_cache\", False)\n\n # Use shared cache if available and caching is enabled\n cached = None\n if use_cache:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n cached = servers_cache.get(server_name) if isinstance(servers_cache, dict) else None\n\n if cached is not None:\n try:\n self.tools = cached[\"tools\"]\n self.tool_names = cached[\"tool_names\"]\n self._tool_cache = cached[\"tool_cache\"]\n server_config_from_value = cached[\"config\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by clearing it and continuing to fetch fresh tools\n msg = f\"Unable to use cached data for MCP Server{server_name}: {e}\"\n await logger.awarning(msg)\n # Clear the corrupted cache entry\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict) and server_name in current_servers_cache:\n current_servers_cache.pop(server_name)\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n else:\n return self.tools, {\"name\": server_name, \"config\": server_config_from_value}\n\n try:\n try:\n from langflow.api.v2.mcp import get_server\n from langflow.services.database.models.user.crud import get_user_by_id\n except ImportError as e:\n msg = (\n \"Langflow MCP server functionality is not available. \"\n \"This feature requires the full Langflow installation.\"\n )\n raise ImportError(msg) from e\n async with session_scope() as db:\n if not self.user_id:\n msg = \"User ID is required for fetching MCP tools.\"\n raise ValueError(msg)\n current_user = await get_user_by_id(db, self.user_id)\n\n # Try to get server config from DB/API\n server_config = await get_server(\n server_name,\n current_user,\n db,\n storage_service=get_storage_service(),\n settings_service=get_settings_service(),\n )\n\n # If get_server returns empty but we have a config, use it\n if not server_config and server_config_from_value:\n server_config = server_config_from_value\n\n if not server_config:\n self.tools = []\n return [], {\"name\": server_name, \"config\": server_config}\n\n _, tool_list, tool_cache = await update_tools(\n server_name=server_name,\n server_config=server_config,\n mcp_stdio_client=self.stdio_client,\n mcp_sse_client=self.sse_client,\n )\n\n self.tool_names = [tool.name for tool in tool_list if hasattr(tool, \"name\")]\n self._tool_cache = tool_cache\n self.tools = tool_list\n\n # Cache the result only if caching is enabled\n if use_cache:\n cache_data = {\n \"tools\": tool_list,\n \"tool_names\": self.tool_names,\n \"tool_cache\": tool_cache,\n \"config\": server_config,\n }\n\n # Safely update the servers cache\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict):\n current_servers_cache[server_name] = cache_data\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout updating tool list: {e!s}\"\n await logger.aexception(msg)\n raise TimeoutError(msg) from e\n except Exception as e:\n msg = f\"Error updating tool list: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return tool_list, {\"name\": server_name, \"config\": server_config}\n\n async def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None) -> dict:\n \"\"\"Toggle the visibility of connection-specific fields based on the selected mode.\"\"\"\n try:\n if field_name == \"tool\":\n try:\n if len(self.tools) == 0:\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout updating tool list: {e!s}\"\n await logger.aexception(msg)\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = \"Timeout on MCP server\"\n else:\n build_config[\"tool\"][\"show\"] = False\n except ValueError:\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = \"Error on MCP Server\"\n else:\n build_config[\"tool\"][\"show\"] = False\n\n if field_value == \"\":\n return build_config\n tool_obj = None\n for tool in self.tools:\n if tool.name == field_value:\n tool_obj = tool\n break\n if tool_obj is None:\n msg = f\"Tool {field_value} not found in available tools: {self.tools}\"\n await logger.awarning(msg)\n return build_config\n await self._update_tool_config(build_config, field_value)\n except Exception as e:\n build_config[\"tool\"][\"options\"] = []\n msg = f\"Failed to update tools: {e!s}\"\n raise ValueError(msg) from e\n else:\n return build_config\n elif field_name == \"mcp_server\":\n if not field_value:\n build_config[\"tool\"][\"show\"] = False\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool_placeholder\"][\"tool_mode\"] = False\n self.remove_non_default_keys(build_config)\n return build_config\n\n build_config[\"tool_placeholder\"][\"tool_mode\"] = True\n\n current_server_name = field_value.get(\"name\") if isinstance(field_value, dict) else field_value\n _last_selected_server = safe_cache_get(self._shared_component_cache, \"last_selected_server\", \"\")\n\n # To avoid unnecessary updates, only proceed if the server has actually changed\n if (_last_selected_server in (current_server_name, \"\")) and build_config[\"tool\"][\"show\"]:\n if current_server_name:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(current_server_name)\n if cached is not None and cached.get(\"tool_names\"):\n cached_tools = cached[\"tool_names\"]\n current_tools = build_config[\"tool\"][\"options\"]\n if current_tools == cached_tools:\n return build_config\n else:\n return build_config\n\n # Determine if \"Tool Mode\" is active by checking if the tool dropdown is hidden.\n is_in_tool_mode = build_config[\"tools_metadata\"][\"show\"]\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", current_server_name)\n\n # Check if tools are already cached for this server before clearing\n cached_tools = None\n if current_server_name:\n use_cache = getattr(self, \"use_cache\", True)\n if use_cache:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(current_server_name)\n if cached is not None:\n try:\n cached_tools = cached[\"tools\"]\n self.tools = cached_tools\n self.tool_names = cached[\"tool_names\"]\n self._tool_cache = cached[\"tool_cache\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by ignoring it\n msg = f\"Unable to use cached data for MCP Server,{current_server_name}: {e}\"\n await logger.awarning(msg)\n cached_tools = None\n\n # Only clear tools if we don't have cached tools for the current server\n if not cached_tools:\n self.tools = [] # Clear previous tools only if no cache\n\n self.remove_non_default_keys(build_config) # Clear previous tool inputs\n\n # Only show the tool dropdown if not in tool_mode\n if not is_in_tool_mode:\n build_config[\"tool\"][\"show\"] = True\n if cached_tools:\n # Use cached tools to populate options immediately\n build_config[\"tool\"][\"options\"] = [tool.name for tool in cached_tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n else:\n # Show loading state only when we need to fetch tools\n build_config[\"tool\"][\"placeholder\"] = \"Loading tools...\"\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n else:\n # Keep the tool dropdown hidden if in tool_mode\n self._not_load_actions = True\n build_config[\"tool\"][\"show\"] = False\n\n elif field_name == \"tool_mode\":\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool\"][\"show\"] = not bool(field_value) and bool(build_config[\"mcp_server\"])\n self.remove_non_default_keys(build_config)\n self.tool = build_config[\"tool\"][\"value\"]\n if field_value:\n self._not_load_actions = True\n else:\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"placeholder\"] = \"Loading tools...\"\n elif field_name == \"tools_metadata\":\n self._not_load_actions = False\n\n except Exception as e:\n msg = f\"Error in update_build_config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return build_config\n\n def get_inputs_for_all_tools(self, tools: list) -> dict:\n \"\"\"Get input schemas for all tools.\"\"\"\n inputs = {}\n for tool in tools:\n if not tool or not hasattr(tool, \"name\"):\n continue\n try:\n flat_schema = flatten_schema(tool.args_schema.schema())\n input_schema = create_input_schema_from_json_schema(flat_schema)\n langflow_inputs = schema_to_langflow_inputs(input_schema)\n inputs[tool.name] = langflow_inputs\n except (AttributeError, ValueError, TypeError, KeyError) as e:\n msg = f\"Error getting inputs for tool {getattr(tool, 'name', 'unknown')}: {e!s}\"\n logger.exception(msg)\n continue\n return inputs\n\n def remove_input_schema_from_build_config(\n self, build_config: dict, tool_name: str, input_schema: dict[list[InputTypes], Any]\n ):\n \"\"\"Remove the input schema for the tool from the build config.\"\"\"\n # Keep only schemas that don't belong to the current tool\n input_schema = {k: v for k, v in input_schema.items() if k != tool_name}\n # Remove all inputs from other tools\n for value in input_schema.values():\n for _input in value:\n if _input.name in build_config:\n build_config.pop(_input.name)\n\n def remove_non_default_keys(self, build_config: dict) -> None:\n \"\"\"Remove non-default keys from the build config.\"\"\"\n for key in list(build_config.keys()):\n if key not in self.default_keys:\n build_config.pop(key)\n\n async def _update_tool_config(self, build_config: dict, tool_name: str) -> None:\n \"\"\"Update tool configuration with proper error handling.\"\"\"\n if not self.tools:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n\n if not tool_name:\n return\n\n tool_obj = next((tool for tool in self.tools if tool.name == tool_name), None)\n if not tool_obj:\n msg = f\"Tool {tool_name} not found in available tools: {self.tools}\"\n self.remove_non_default_keys(build_config)\n build_config[\"tool\"][\"value\"] = \"\"\n await logger.awarning(msg)\n return\n\n try:\n # Store current values before removing inputs\n current_values = {}\n for key, value in build_config.items():\n if key not in self.default_keys and isinstance(value, dict) and \"value\" in value:\n current_values[key] = value[\"value\"]\n\n # Get all tool inputs and remove old ones\n input_schema_for_all_tools = self.get_inputs_for_all_tools(self.tools)\n self.remove_input_schema_from_build_config(build_config, tool_name, input_schema_for_all_tools)\n\n # Get and validate new inputs\n self.schema_inputs = await self._validate_schema_inputs(tool_obj)\n if not self.schema_inputs:\n msg = f\"No input parameters to configure for tool '{tool_name}'\"\n await logger.ainfo(msg)\n return\n\n # Add new inputs to build config\n for schema_input in self.schema_inputs:\n if not schema_input or not hasattr(schema_input, \"name\"):\n msg = \"Invalid schema input detected, skipping\"\n await logger.awarning(msg)\n continue\n\n try:\n name = schema_input.name\n input_dict = schema_input.to_dict()\n input_dict.setdefault(\"value\", None)\n input_dict.setdefault(\"required\", True)\n\n build_config[name] = input_dict\n\n # Preserve existing value if the parameter name exists in current_values\n if name in current_values:\n build_config[name][\"value\"] = current_values[name]\n\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error processing schema input {schema_input}: {e!s}\"\n await logger.aexception(msg)\n continue\n except ValueError as e:\n msg = f\"Schema validation error for tool {tool_name}: {e!s}\"\n await logger.aexception(msg)\n self.schema_inputs = []\n return\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error updating tool config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n async def build_output(self) -> DataFrame:\n \"\"\"Build output with improved error handling and validation.\"\"\"\n try:\n self.tools, _ = await self.update_tool_list()\n if self.tool != \"\":\n # Set session context for persistent MCP sessions using Langflow session ID\n session_context = self._get_session_context()\n if session_context:\n self.stdio_client.set_session_context(session_context)\n self.sse_client.set_session_context(session_context)\n\n exec_tool = self._tool_cache[self.tool]\n tool_args = self.get_inputs_for_all_tools(self.tools)[self.tool]\n kwargs = {}\n for arg in tool_args:\n value = getattr(self, arg.name, None)\n if value is not None:\n if isinstance(value, Message):\n kwargs[arg.name] = value.text\n else:\n kwargs[arg.name] = value\n\n unflattened_kwargs = maybe_unflatten_dict(kwargs)\n\n output = await exec_tool.coroutine(**unflattened_kwargs)\n\n tool_content = []\n for item in output.content:\n item_dict = item.model_dump()\n tool_content.append(item_dict)\n return DataFrame(data=tool_content)\n return DataFrame(data=[{\"error\": \"You must select a tool\"}])\n except Exception as e:\n msg = f\"Error in build_output: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n def _get_session_context(self) -> str | None:\n \"\"\"Get the Langflow session ID for MCP session caching.\"\"\"\n # Try to get session ID from the component's execution context\n if hasattr(self, \"graph\") and hasattr(self.graph, \"session_id\"):\n session_id = self.graph.session_id\n # Include server name to ensure different servers get different sessions\n server_name = \"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\", \"\")\n elif mcp_server:\n server_name = str(mcp_server)\n return f\"{session_id}_{server_name}\" if session_id else None\n return None\n\n async def _get_tools(self):\n \"\"\"Get cached tools or update if necessary.\"\"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n if not self._not_load_actions:\n tools, _ = await self.update_tool_list(mcp_server)\n return tools\n return []\n" + }, + "mcp_server": { + "_input_type": "McpInput", + "advanced": false, + "display_name": "MCP Server", + "dynamic": false, + "info": "Select the MCP Server that will be used by this component", + "name": "mcp_server", + "placeholder": "", + "real_time_refresh": true, + "required": false, + "show": true, + "title_case": false, + "trace_as_metadata": true, + "type": "mcp", + "value": { + "config": { + "args": [ + "mcp-proxy", + "--headers", + "x-api-key", + "sk-lq7nQIiX4jbYTIOGH7YG9z46E0IW1i-FSvn_hkcg2xE", + "http://localhost:7860/api/v1/mcp/project/304fb921-38e4-4763-b223-832a3e3546e0/sse" + ], + "command": "uvx" + }, + "name": "lf-starter_project" + } + }, + "tool": { + "_input_type": "DropdownInput", + "advanced": false, + "combobox": false, + "dialog_inputs": {}, + "display_name": "Tool", + "dynamic": false, + "external_options": {}, + "info": "Select the tool to execute", + "name": "tool", + "options": [ + "opensearch_url_ingestion_flow" + ], + "options_metadata": [], + "placeholder": "", + "real_time_refresh": true, + "required": true, + "show": false, + "title_case": false, + "toggle": false, + "tool_mode": false, + "trace_as_metadata": true, + "type": "str", + "value": "" + }, + "tool_placeholder": { + "_input_type": "MessageTextInput", + "advanced": false, + "display_name": "Tool Placeholder", + "dynamic": false, + "info": "Placeholder for the tool", + "input_types": [ + "Message" + ], + "list": false, + "list_add_label": "Add More", + "load_from_db": false, + "name": "tool_placeholder", + "placeholder": "", + "required": false, + "show": false, + "title_case": false, + "tool_mode": true, + "trace_as_input": true, + "trace_as_metadata": true, + "type": "str", + "value": "" + }, + "tools_metadata": { + "_input_type": "ToolsInput", + "advanced": false, + "display_name": "Actions", + "dynamic": false, + "info": "Modify tool names and descriptions to help agents understand when to use each tool.", + "is_list": true, + "list_add_label": "Add More", + "name": "tools_metadata", + "placeholder": "", + "real_time_refresh": true, + "required": false, + "show": true, + "title_case": false, + "tool_mode": false, + "trace_as_metadata": true, + "type": "tools", + "value": [ + { + "args": { + "input_value": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Message to be passed as input.", + "title": "Input Value" + } + }, + "description": "This flow is to ingest the URL to open search.", + "display_description": "This flow is to ingest the URL to open search.", + "display_name": "opensearch_url_ingestion_flow", + "name": "opensearch_url_ingestion_flow", + "readonly": false, + "status": true, + "tags": [ + "opensearch_url_ingestion_flow" + ] + } + ] + }, + "use_cache": { + "_input_type": "BoolInput", + "advanced": true, + "display_name": "Use Cached Server", + "dynamic": false, + "info": "Enable caching of MCP Server and tools to improve performance. Disable to always fetch fresh tools and server updates.", + "list": false, + "list_add_label": "Add More", + "name": "use_cache", + "placeholder": "", + "required": false, + "show": true, + "title_case": false, + "tool_mode": false, + "trace_as_metadata": true, + "type": "bool", + "value": false + } + }, + "tool_mode": true + }, + "showNode": true, + "type": "MCP" + }, + "id": "MCP-7EY21", + "measured": { + "height": 284, + "width": 320 + }, + "position": { + "x": 675.7137923419156, + "y": 878.6218422334763 + }, + "selected": false, + "type": "genericNode" } ], "viewport": { - "x": -149.48015964664273, + "x": -237.0727605845459, "y": 154.6885920024542, "zoom": 0.602433700773958 } @@ -2563,8 +2836,8 @@ "endpoint_name": null, "id": "1098eea1-6649-4e1d-aed1-b77249fb8dd0", "is_component": false, - "last_tested_version": "1.6.3.dev0", - "name": "OpenRAG OpenSearch Agent", + "last_tested_version": "1.6.0", + "name": "OpenRAG Open Search Agent", "tags": [ "assistants", "agents" diff --git a/flows/openrag_url_mcp.json b/flows/openrag_url_mcp.json index 69dbc85d..9cab0fed 100644 --- a/flows/openrag_url_mcp.json +++ b/flows/openrag_url_mcp.json @@ -232,6 +232,7 @@ }, { "animated": false, + "className": "", "data": { "sourceHandle": { "dataType": "EmbeddingModel", @@ -733,6 +734,10 @@ { "key": "owner_email", "value": "OWNER_EMAIL" + }, + { + "key": "connector_type", + "value": "CONNECTOR_TYPE_URL" } ] }, @@ -1808,7 +1813,7 @@ ], "frozen": false, "icon": "table", - "last_updated": "2025-10-03T20:31:36.023Z", + "last_updated": "2025-10-06T17:46:55.068Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -2224,7 +2229,7 @@ ], "frozen": false, "icon": "table", - "last_updated": "2025-10-03T20:31:36.025Z", + "last_updated": "2025-10-06T17:46:55.069Z", "legacy": false, "lf_version": "1.6.0", "metadata": { @@ -2897,7 +2902,7 @@ ], "frozen": false, "icon": "table", - "last_updated": "2025-10-03T20:31:36.026Z", + "last_updated": "2025-10-06T17:46:55.069Z", "legacy": false, "metadata": { "code_hash": "b4d6b19b6eef", @@ -3310,7 +3315,7 @@ ], "frozen": false, "icon": "binary", - "last_updated": "2025-10-03T20:31:47.177Z", + "last_updated": "2025-10-06T17:46:54.996Z", "legacy": false, "metadata": { "code_hash": "8607e963fdef", @@ -3595,17 +3600,17 @@ } ], "viewport": { - "x": -407.1633937626607, - "y": -577.5291936220412, - "zoom": 0.5347553210574026 + "x": -538.2311610019549, + "y": -337.3313239657308, + "zoom": 0.45546556043892106 } }, "description": "This flow is to ingest the URL to open search.", "endpoint_name": null, - "mcp_enabled": true, "id": "72c3d17c-2dac-4a73-b48a-6518473d7830", + "mcp_enabled": true, "is_component": false, - "last_tested_version": "1.6.0", + "last_tested_version": "1.6.3.dev1", "name": "OpenSearch URL Ingestion Flow", "tags": [ "openai", diff --git a/src/services/auth_service.py b/src/services/auth_service.py index 6b19f77a..f58997ac 100644 --- a/src/services/auth_service.py +++ b/src/services/auth_service.py @@ -296,11 +296,16 @@ class AuthService: try: if self.langflow_mcp_service and isinstance(jwt_token, str) and jwt_token.strip(): global_vars = {"JWT": jwt_token} + global_vars["CONNECTOR_TYPE_URL"] = "url" if user_info: if user_info.get("id"): global_vars["OWNER"] = user_info.get("id") if user_info.get("name"): - global_vars["OWNER_NAME"] = user_info.get("name") + # OWNER_NAME may contain spaces, which can cause issues in headers. + # Alternative: URL-encode the owner name to preserve spaces and special characters. + owner_name = user_info.get("name") + if owner_name: + global_vars["OWNER_NAME"] = str(f"\"{owner_name}\"") if user_info.get("email"): global_vars["OWNER_EMAIL"] = user_info.get("email") diff --git a/src/tui/_assets/docker-compose-cpu.yml b/src/tui/_assets/docker-compose-cpu.yml index 1086737b..4a1125f8 100644 --- a/src/tui/_assets/docker-compose-cpu.yml +++ b/src/tui/_assets/docker-compose-cpu.yml @@ -105,6 +105,7 @@ services: - OWNER_NAME=None - OWNER_EMAIL=None - CONNECTOR_TYPE=system + - CONNECTOR_TYPE_URL=url - OPENRAG-QUERY-FILTER="{}" - OPENSEARCH_PASSWORD=${OPENSEARCH_PASSWORD} - FILENAME=None diff --git a/src/tui/_assets/docker-compose.yml b/src/tui/_assets/docker-compose.yml index 32b72c65..6cac6506 100644 --- a/src/tui/_assets/docker-compose.yml +++ b/src/tui/_assets/docker-compose.yml @@ -105,6 +105,7 @@ services: - OWNER_NAME=None - OWNER_EMAIL=None - CONNECTOR_TYPE=system + - CONNECTOR_TYPE_URL=url - OPENRAG-QUERY-FILTER="{}" - OPENSEARCH_PASSWORD=${OPENSEARCH_PASSWORD} - FILENAME=None