From fa249352a22d628b45e7b9f141575c95866f76ec Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Tue, 20 Oct 2026 15:53:19 +0000 Subject: [PATCH 01/32] chore: auto-bake note keys and regenerate backend locales/en.json [skip ci] --- .../starter_projects/Basic Prompting.json | 2 +- .../Custom Component Generator.json | 2 +- .../starter_projects/Document Q&A.json | 2 +- .../Research Translation Loop.json | 2 +- src/backend/base/langflow/locales/en.json | 41 ++++++++----------- 5 files changed, 22 insertions(+), 27 deletions(-) diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Basic Prompting.json b/src/backend/base/langflow/initial_setup/starter_projects/Basic Prompting.json index cc1e9967e9..4eadb21a36 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Basic Prompting.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Basic Prompting.json @@ -524,7 +524,7 @@ "description": "# πŸ“– README\nThis template demonstrates a standard chat flow with additional instructions provided by a prompt. Prompts provide instructions and inputs for a Large Language Model (LLM) beyond the standard user-provided chat input. In this example, the prompt describes the LLM's role and persona.\n\n## Quick start\n1. Configure your **Model Provider** with your API credentials.\n2. Open the **Playground** to start the chat and run the flow.\n\n## Next steps\nChange the prompt template, model, or model settings, such as **Temperature**, and then see how the responses change with these different inputs.\nπŸ’‘ Some component settings are hidden by default; to view all settings click **Controls** in each component's header menu.\nπŸ’‘ You can use curly braces to create variables in your template, such as `{variable}`. These can be populated from other components, with OpenXFlow global variables, or at runtime.", "display_name": "", "documentation": "", - "i18n_key": "template_notes.basic_prompting.bd8ff52b", + "i18n_key": "template_notes.basic_prompting.ee8855e9", "template": { "backgroundColor": "neutral" } diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Custom Component Generator.json b/src/backend/base/langflow/initial_setup/starter_projects/Custom Component Generator.json index fbb2a10b68..4435716610 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Custom Component Generator.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Custom Component Generator.json @@ -1811,7 +1811,7 @@ "description": "# 🧩 Custom Component Generator\n\nDescribe the component you want and an **Agent** writes an OpenXFlow custom Component (Python). It pulls OpenXFlow's reference code on demand and remembers the conversation so you can iterate.\n\n## Setup\n1. **Agent** - add your LLM key (any provider in the Model dropdown).\n2. **URL** and **Memory** tools need no key.\n\n## How it works\n1. **Chat Input β†’ Agent** - your component request.\n2. The Agent uses the **URL** tool to fetch OpenXFlow reference code (base class, examples) when it needs the exact conventions, and **Memory** to recall earlier turns.\n3. **Agent β†’ Chat Output** - the generated Python. Ask follow-ups to refine it.\n\n## Run\nOpen **Playground**, describe the component you need, and run.", "display_name": "", "documentation": "", - "i18n_key": "template_notes.custom_component_generator.b1d3ed28", + "i18n_key": "template_notes.custom_component_generator.5073eb43", "template": { "backgroundColor": "blue" } diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json b/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json index db6d83d37d..9651fd3e6c 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json @@ -2098,7 +2098,7 @@ "description": "# πŸ“„ Document Q&A\n\nAsk questions about your documents and get answers grounded in their content,\nusing a Knowledge Base for retrieval.\n\n## Setup\n1. **Knowledge Base** β€” add your documents to a Knowledge Base first (via the\n **Knowledge** module in OpenXFlow, or a separate ingestion flow). This flow\n only **retrieves** β€” it does not ingest.\n2. **Knowledge (Retrieve)** β€” select the Knowledge Base that holds your documents.\n3. **Agent** β€” add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input β†’ Knowledge (Retrieve)** β€” your question pulls the most relevant\n passages from the Knowledge Base.\n2. **Parser β†’ Prompt** β€” the retrieved passages become the context, combined with\n your question.\n3. **Agent β†’ Chat Output** β€” the Agent answers using **only** the retrieved\n context (it won't make things up beyond your documents).\n\n## Run\nMake sure your Knowledge Base is populated, then type a question in the Playground.", "display_name": "", "documentation": "", - "i18n_key": "template_notes.document_q_a.0e21c05f", + "i18n_key": "template_notes.document_q_a.f7a494a0", "template": { "backgroundColor": "blue" } diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json b/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json index 911707acf0..d1a07e9a72 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json @@ -921,7 +921,7 @@ "description": "# πŸ“– README\nThis template translates research paper summaries on ArXiv into Portuguese and summarizes them. \n Using **OpenXFlow’s looping mechanism**, the template iterates through multiple research papers, translates them with the **OpenAI** model component, and outputs an aggregated version of all translated papers. \n\n## Quick start\n1. Configure your **Model Provider** with your API credentials. \n2. In the **Playground**, enter a query related to a research topic (for example, β€œQuantum Computing Advancements”). \n\n The flow fetches a list of research papers from ArXiv matching the query. Each paper in the retrieved list is processed one-by-one using the OpenXFlow **Loop component**. \n\n The abstract of each paper is translated into Portuguese by the **OpenAI** model component. \n\n Once all papers are translated, the system aggregates them into a **single structured output**.", "display_name": "", "documentation": "", - "i18n_key": "template_notes.research_translation_loop.60bb882f", + "i18n_key": "template_notes.research_translation_loop.c0723e26", "template": {} }, "type": "note" diff --git a/src/backend/base/langflow/locales/en.json b/src/backend/base/langflow/locales/en.json index de4d33f97b..97bc7686ab 100644 --- a/src/backend/base/langflow/locales/en.json +++ b/src/backend/base/langflow/locales/en.json @@ -95,7 +95,7 @@ "components.altermetadata.outputs.data.info.d99d8586": "List of Input objects each with added Metadata", "components.altermetadata.outputs.dataframe.display_name.16d1c905": "Table", "components.altermetadata.outputs.dataframe.info.64cf6cf1": "Data objects as a DataFrame, with metadata as columns", - "components.altkagent.description.f5f1d3c5": "ALTK was retired in OpenXFlow 1.13.0. Replace this node with the Agent component.", + "components.altkagent.description.4da568f4": "ALTK was retired in OpenXFlow 1.13.0. Replace this node with the Agent component.", "components.altkagent.display_name.73afda4a": "ALTK Agent", "components.altkagent.inputs.add_calculator_tool.display_name.668b8045": "Calculator", "components.altkagent.inputs.add_calculator_tool.info.a45cfd4c": "If true, adds a zero-config arithmetic calculator tool to the agent (safe: only +, -, *, /, ** operators via AST).", @@ -128,6 +128,8 @@ "components.altkagent.inputs.model.placeholder.56590d8e": "Setup Provider", "components.altkagent.inputs.n_messages.display_name.3105e924": "Number of Chat History Messages", "components.altkagent.inputs.n_messages.info.39873fa8": "Number of chat history messages to retrieve.", + "components.altkagent.inputs.openai_compatible_base_url.display_name.c85ef531": "OpenAI-Compatible API URL", + "components.altkagent.inputs.openai_compatible_base_url.info.ccfff607": "Overrides the configured API endpoint for OpenAI-compatible providers.", "components.altkagent.inputs.output_schema.display_name.21e3cb67": "Output Schema", "components.altkagent.inputs.output_schema.info.3b7a7b80": "Schema Validation: Define the structure and data types for structured output. No validation if no output schema.", "components.altkagent.inputs.project_id.display_name.c7feb8db": "watsonx Project ID", @@ -1278,6 +1280,8 @@ "components.file.inputs.storage_location.display_name.6b1f9093": "Storage Location", "components.file.inputs.storage_location.info.713b4100": "Choose where to read the file from.", "components.file.inputs.storage_location.placeholder.3c82033f": "Select Location", + "components.file.inputs.text_encoding.display_name.ab5ad1dc": "Text Encoding", + "components.file.inputs.text_encoding.info.6a86b444": "Leave blank to detect text encoding automatically. If accents or punctuation look incorrect, enter the file's encoding, such as cp1252 or utf-8. Applies to standard text parsing and CSV, not PDF, DOCX, or the Advanced Parser. The selected encoding is shown in logs and output metadata.", "components.file.inputs.use_multithreading.display_name.a32391d0": "[Deprecated] Use Multithreading", "components.file.inputs.use_multithreading.info.e1193371": "Set 'Processing Concurrency' greater than 1 to enable multithreading.", "components.file.outputs.message.display_name.960c8537": "Raw Content", @@ -1849,7 +1853,7 @@ "components.jsontodata.inputs.json_string.display_name.560082ac": "JSON String", "components.jsontodata.inputs.json_string.info.5786aa41": "Enter a valid JSON string (object or array) to convert to a Data object or list of Data objects", "components.jsontodata.outputs.data.display_name.db1a21a0": "JSON", - "components.knowledge.description.a1a10675": "Ingest into or retrieve from an OpenXFlow knowledge base.", + "components.knowledge.description.87015105": "Ingest into or retrieve from an OpenXFlow knowledge base.", "components.knowledge.display_name.dcb3e1c0": "Knowledge", "components.knowledge.inputs.allow_duplicates.display_name.c5507652": "Allow Duplicates", "components.knowledge.inputs.allow_duplicates.info.4f0b36ee": "Allow duplicate rows in the knowledge base", @@ -1910,7 +1914,7 @@ "components.knowledgebase.inputs.top_k.info.dc908edc": "Number of top results to return from the knowledge base.", "components.knowledgebase.outputs.retrieve_data.display_name.219c4a6c": "Results", "components.knowledgebase.outputs.retrieve_data.info.791a5874": "Returns the data from the selected knowledge base.", - "components.knowledgeingestion.description.4a211a27": "Create or update knowledge in OpenXFlow.", + "components.knowledgeingestion.description.4248eab8": "Create or update knowledge in OpenXFlow.", "components.knowledgeingestion.display_name.89439d2d": "Knowledge Ingestion", "components.knowledgeingestion.inputs.allow_duplicates.display_name.c5507652": "Allow Duplicates", "components.knowledgeingestion.inputs.allow_duplicates.info.4f0b36ee": "Allow duplicate rows in the knowledge base", @@ -2073,12 +2077,12 @@ "components.mcptools.inputs.verify_ssl.display_name.e9ee31b8": "Verify SSL Certificate", "components.mcptools.inputs.verify_ssl.info.63deba67": "Enable SSL certificate verification for HTTPS connections. Disable only for development/testing with self-signed certificates.", "components.mcptools.outputs.response.display_name.9061383b": "Response", - "components.memory.description.31449424": "Stores or retrieves stored chat messages from OpenXFlow tables or an external memory.", + "components.memory.description.5e6e6979": "Stores or retrieves stored chat messages from OpenXFlow tables or an external memory.", "components.memory.display_name.64169537": "Message History", "components.memory.inputs.context_id.display_name.2051cde7": "Context ID", "components.memory.inputs.context_id.info.55bf9c83": "The context ID of the chat. Adds an extra layer to the local memory.", "components.memory.inputs.memory.display_name.c7a62e64": "External Memory", - "components.memory.inputs.memory.info.c2d79141": "Retrieve messages from an external memory. If empty, it will use the OpenXFlow tables.", + "components.memory.inputs.memory.info.8d537dfe": "Retrieve messages from an external memory. If empty, it will use the OpenXFlow tables.", "components.memory.inputs.message.display_name.2f77668a": "Message", "components.memory.inputs.message.info.212d7fd0": "The chat message to be stored.", "components.memory.inputs.mode.display_name.5e23ec6a": "Mode", @@ -3059,7 +3063,7 @@ "components.slackonmessagetrigger.inputs.channels.display_name.847bd826": "Channel IDs", "components.slackonmessagetrigger.inputs.channels.info.d3adb3e5": "Only fire for these conversations, by ID (for example C0123456789, from the channel's details). Leave empty for every conversation the app can see.", "components.slackonmessagetrigger.inputs.connection.display_name.335f10dc": "Slack Connection", - "components.slackonmessagetrigger.inputs.connection.info.3126fcb2": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", + "components.slackonmessagetrigger.inputs.connection.info.2be79dca": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", "components.slackonmessagetrigger.inputs.conversation_types.display_name.9a6aa1a0": "Conversation types", "components.slackonmessagetrigger.inputs.conversation_types.info.802d162a": "channel: public channels. group: private channels. im: direct messages with the app. mpim: group direct messages.", "components.slackonmessagetrigger.inputs.event_payload.display_name.4e1f49a9": "Event", @@ -3079,7 +3083,7 @@ "components.slackonreactiontrigger.inputs.channels.display_name.847bd826": "Channel IDs", "components.slackonreactiontrigger.inputs.channels.info.d3adb3e5": "Only fire for these conversations, by ID (for example C0123456789, from the channel's details). Leave empty for every conversation the app can see.", "components.slackonreactiontrigger.inputs.connection.display_name.335f10dc": "Slack Connection", - "components.slackonreactiontrigger.inputs.connection.info.3126fcb2": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", + "components.slackonreactiontrigger.inputs.connection.info.2be79dca": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", "components.slackonreactiontrigger.inputs.emoji.display_name.61ad8976": "Emoji", "components.slackonreactiontrigger.inputs.emoji.info.4c80a33d": "Emoji names without colons, for example 'rocket'. Leave empty for any reaction.", "components.slackonreactiontrigger.inputs.event_payload.display_name.4e1f49a9": "Event", @@ -3250,10 +3254,10 @@ "components.sqlgenerator.inputs.top_k.display_name.74fc1150": "Top K", "components.sqlgenerator.inputs.top_k.info.481a29ee": "The number of results per select statement to return.", "components.sqlgenerator.outputs.text.display_name.2f77668a": "Message", - "components.storemessage.description.786ee729": "Stores a chat message or text into OpenXFlow tables or an external memory.", + "components.storemessage.description.ab58b04c": "Stores a chat message or text into OpenXFlow tables or an external memory.", "components.storemessage.display_name.a8c1d3a3": "Message Store", "components.storemessage.inputs.memory.display_name.c7a62e64": "External Memory", - "components.storemessage.inputs.memory.info.1bd5d94f": "The external memory to store the message. If empty, it will use the OpenXFlow tables.", + "components.storemessage.inputs.memory.info.bdde8480": "The external memory to store the message. If empty, it will use the OpenXFlow tables.", "components.storemessage.inputs.message.display_name.2f77668a": "Message", "components.storemessage.inputs.message.info.212d7fd0": "The chat message to be stored.", "components.storemessage.inputs.sender.display_name.c299f9b8": "Sender", @@ -3671,13 +3675,13 @@ "starter_flows.twitter_thread_generator.name": "Twitter Thread Generator", "starter_flows.vector_store_rag.description": "Load your data for chat context with Retrieval Augmented Generation.", "starter_flows.vector_store_rag.name": "Vector Store RAG", - "template_notes.basic_prompting.bd8ff52b": "# πŸ“– README\nThis template demonstrates a standard chat flow with additional instructions provided by a prompt. Prompts provide instructions and inputs for a Large Language Model (LLM) beyond the standard user-provided chat input. In this example, the prompt describes the LLM's role and persona.\n\n## Quick start\n1. Configure your **Model Provider** with your API credentials.\n2. Open the **Playground** to start the chat and run the flow.\n\n## Next steps\nChange the prompt template, model, or model settings, such as **Temperature**, and then see how the responses change with these different inputs.\nπŸ’‘ Some component settings are hidden by default; to view all settings click **Controls** in each component's header menu.\nπŸ’‘ You can use curly braces to create variables in your template, such as `{variable}`. These can be populated from other components, with OpenXFlow global variables, or at runtime.", "template_notes.basic_prompting.cd87cff6": "### Configure your Model Provider", + "template_notes.basic_prompting.ee8855e9": "# πŸ“– README\nThis template demonstrates a standard chat flow with additional instructions provided by a prompt. Prompts provide instructions and inputs for a Large Language Model (LLM) beyond the standard user-provided chat input. In this example, the prompt describes the LLM's role and persona.\n\n## Quick start\n1. Configure your **Model Provider** with your API credentials.\n2. Open the **Playground** to start the chat and run the flow.\n\n## Next steps\nChange the prompt template, model, or model settings, such as **Temperature**, and then see how the responses change with these different inputs.\nπŸ’‘ Some component settings are hidden by default; to view all settings click **Controls** in each component's header menu.\nπŸ’‘ You can use curly braces to create variables in your template, such as `{variable}`. These can be populated from other components, with OpenXFlow global variables, or at runtime.", "template_notes.blog_writer.23fde7d9": "# ✍️ Blog Writer\n\nGive it a **URL** and it writes a blog post grounded in that page. An **Agent** fetches the page itself using the **URL component as a tool** (Fetch Content), then writes the article in Markdown. Rebuilt on the new Agent core component β€” no separate parsing step.\n\n## Setup\n1. **Agent** β€” pick your **Model Provider** and paste its API key (OpenAI, Anthropic, Google, Groq, etc. β€” any option in the Model dropdown).\n2. **URL tool** β€” already wired into the Agent's **Tools**; no setup needed. Keep **Depth = 1** so it fetches just the page you give it (higher depth follows links and costs more time/tokens).\n\n## Use\n1. Open the **Playground**.\n2. Paste a **source URL**, e.g. `https://docs.langflow.org/`.\n3. The Agent fetches that page through the URL tool and returns a finished blog post.\n\n> If nothing runs on an empty task, type `Write the blog post from the source URL.` in the Agent's **Input** field.\n\n## How it works\n1. **Chat Input** β€” you enter the source URL.\n2. **Prompt Template `{URL}`** β€” injects the URL into the Agent's instructions.\n3. **Agent (+ URL tool)** β€” calls the URL **Fetch Content** tool to retrieve the page, then writes a Markdown post grounded only in what it fetched (no invented facts).\n4. **Chat Output** β€” the finished blog post.", "template_notes.content_aggregator.cbed07a7": "# πŸ“° Content Aggregator\n\nSearch the news on a topic, summarize each article, and compose a clean aggregated digest - then save it to a file.\n\n## Setup\n1. **Agent** and **Batch Run** - add your LLM key on each (any provider in the Model dropdown).\n2. **Web Search** - needs no key. Set its **Search Mode** to *News* (already selected); optionally set a **Topic** (Business, Technology, ...).\n3. **Save to File** - choose where to write the digest.\n\n## How it works\n1. **Chat Input -> Web Search (News)** - finds recent articles on your topic (returns title, link, date, summary).\n2. **Batch Run** - summarizes every article in one line.\n3. **Parser -> Agent** - the summaries are formatted and the Agent curates them into a clean digest.\n4. **Chat Output -> Save to File** - the digest is shown and saved.\n\n## Run\nType a topic (e.g. \"artificial intelligence\") and run.", - "template_notes.custom_component_generator.b1d3ed28": "# 🧩 Custom Component Generator\n\nDescribe the component you want and an **Agent** writes an OpenXFlow custom Component (Python). It pulls OpenXFlow's reference code on demand and remembers the conversation so you can iterate.\n\n## Setup\n1. **Agent** - add your LLM key (any provider in the Model dropdown).\n2. **URL** and **Memory** tools need no key.\n\n## How it works\n1. **Chat Input β†’ Agent** - your component request.\n2. The Agent uses the **URL** tool to fetch OpenXFlow reference code (base class, examples) when it needs the exact conventions, and **Memory** to recall earlier turns.\n3. **Agent β†’ Chat Output** - the generated Python. Ask follow-ups to refine it.\n\n## Run\nOpen **Playground**, describe the component you need, and run.", + "template_notes.custom_component_generator.5073eb43": "# 🧩 Custom Component Generator\n\nDescribe the component you want and an **Agent** writes an OpenXFlow custom Component (Python). It pulls OpenXFlow's reference code on demand and remembers the conversation so you can iterate.\n\n## Setup\n1. **Agent** - add your LLM key (any provider in the Model dropdown).\n2. **URL** and **Memory** tools need no key.\n\n## How it works\n1. **Chat Input β†’ Agent** - your component request.\n2. The Agent uses the **URL** tool to fetch OpenXFlow reference code (base class, examples) when it needs the exact conventions, and **Memory** to recall earlier turns.\n3. **Agent β†’ Chat Output** - the generated Python. Ask follow-ups to refine it.\n\n## Run\nOpen **Playground**, describe the component you need, and run.", "template_notes.deep_research_agent.195c6a9d": "# πŸ”Ž Deeply Research Agent\n\nAsk a research question and get a thorough, synthesized answer. Three Agents work together: one plans, one researches the web, and one writes up the findings.\n\n## Setup\n1. Add your LLM key on each of the three **Agents**.\n2. **Web Search** - no key needed (DuckDuckGo / Google News).\n\n## How it works\n1. **Chat Input β†’ Planner Agent** - drafts a plan / initial take on the question.\n2. **Search Agent (+ Web Search)** - researches the topic on the web.\n3. **Synthesizer Agent** - combines the question, plan, and search results into a final answer.\n4. **β†’ Chat Output** - the researched answer.\n\n## Run\nOpen **Playground**, ask a research question, and run.", - "template_notes.document_q_a.0e21c05f": "# πŸ“„ Document Q&A\n\nAsk questions about your documents and get answers grounded in their content,\nusing a Knowledge Base for retrieval.\n\n## Setup\n1. **Knowledge Base** β€” add your documents to a Knowledge Base first (via the\n **Knowledge** module in OpenXFlow, or a separate ingestion flow). This flow\n only **retrieves** β€” it does not ingest.\n2. **Knowledge (Retrieve)** β€” select the Knowledge Base that holds your documents.\n3. **Agent** β€” add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input β†’ Knowledge (Retrieve)** β€” your question pulls the most relevant\n passages from the Knowledge Base.\n2. **Parser β†’ Prompt** β€” the retrieved passages become the context, combined with\n your question.\n3. **Agent β†’ Chat Output** β€” the Agent answers using **only** the retrieved\n context (it won't make things up beyond your documents).\n\n## Run\nMake sure your Knowledge Base is populated, then type a question in the Playground.", + "template_notes.document_q_a.f7a494a0": "# πŸ“„ Document Q&A\n\nAsk questions about your documents and get answers grounded in their content,\nusing a Knowledge Base for retrieval.\n\n## Setup\n1. **Knowledge Base** β€” add your documents to a Knowledge Base first (via the\n **Knowledge** module in OpenXFlow, or a separate ingestion flow). This flow\n only **retrieves** β€” it does not ingest.\n2. **Knowledge (Retrieve)** β€” select the Knowledge Base that holds your documents.\n3. **Agent** β€” add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input β†’ Knowledge (Retrieve)** β€” your question pulls the most relevant\n passages from the Knowledge Base.\n2. **Parser β†’ Prompt** β€” the retrieved passages become the context, combined with\n your question.\n3. **Agent β†’ Chat Output** β€” the Agent answers using **only** the retrieved\n context (it won't make things up beyond your documents).\n\n## Run\nMake sure your Knowledge Base is populated, then type a question in the Playground.", "template_notes.financial_report_parser.5972260b": "# πŸ“Š Financial Report Parser (Agent)\n\nUpload a financial report and an **Agent** extracts the key metrics (Gross Profit, EBITDA, Net Income, ...) into a structured table. Rebuilt on the **Agent** core component using its built-in **structured output** - replaces the standalone Structured Output component.\n\n## Setup\n1. **File** - upload your financial report (PDF, DOCX, XLSX, TXT).\n2. **Agent** - pick your **Model Provider** + API key. Edit the **Output Schema** to change which metrics are extracted.\n\n## How it works\n1. **File β†’ Agent** - the Agent reads the report and extracts the schema fields.\n2. **Agent (Structured Response) β†’ Parser** - turns the extracted JSON into readable text.\n3. **Parser β†’ Chat Output** - shows the structured metrics.\n\n## Run\nOpen **Playground**, upload the report, and run. The Agent returns only values found in the document (a field stays blank when the metric isn't present).", "template_notes.hybrid_search_rag.9e93986f": "# πŸ”Ž 2) Search\n\n1. **Chat Input** - your question goes to Astra DB as the semantic query.\n2. **Agent** - extracts keywords from the question for the lexical search.\n3. **Astra DB** - combines + reranks semantic and lexical, returns the top passages.\n4. **Parser -> Chat Output** - the passages as text.\n\nRun from the Playground (or β–Ά on Chat Output) after the collection is populated.", "template_notes.hybrid_search_rag.bc1f15c7": "# πŸ“₯ 1) Ingestion - run once\n\nPopulate the collection so Search has data.\n\n1. **File** - upload your document (e.g. the sample `.md`).\n2. **Split Text** - chunks it (size 1000, overlap 100).\n3. **Astra DB** - set Token + Database + Collection, then press β–Ά on this Astra DB node to write the chunks (Vectorize embeds + lexically indexes on write).", @@ -3691,7 +3695,7 @@ "template_notes.multi_agent_flow.5a71f2d1": "# 🧩 Multi Agent flow\n\nA three-step chain of specialist Agents that turns a market trend into a development plan. Each Agent passes its result to the next.\n\n## Setup\nAdd your LLM key on each **Agent** (any provider in the Model dropdown).\n\n## How it works\n1. **Innovator** - reads the market trend and proposes a product concept.\n2. **Analyst** - evaluates the concept's market potential, feasibility, and risks.\n3. **Product Manager** - turns it into a simple development and launch plan.\n\n## Run\nType a market trend in the chat (a default one is provided) and run; the final plan appears in Chat Output.", "template_notes.portfolio_website_code_generator.c217664c": "# πŸ’» Portfolio Website Code Generator\n\nUpload your resume and an **Agent** generates a single-page portfolio website (HTML + CSS).\n\n## Setup\n1. **File** - upload your resume (PDF or TXT).\n2. **Agent** - add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **File β†’ Agent** - the Agent reads your resume.\n2. **Agent** - extracts your details and writes the portfolio site (header, about, experience, projects, skills, contact) as HTML/CSS.\n3. **Agent β†’ Chat Output** - copy the generated code.\n\n## Run\nUpload your resume, open **Playground**, and run. Tip: a clearly formatted resume (headings, bullet points) produces the best result.", "template_notes.price_deal_finder.6de222f2": "# πŸ’° Price Deal Finder\n\nAn **Agent** that hunts for the best price on a product: it searches the web (and coupon news), reads listing pages, does the math, and remembers what you're tracking - returning the deals as a structured table.\n\n## Setup\n1. **Agent** - add your LLM key. Its tools are Web Search, URL, Calculator, and Memory.\n2. **Web Search** / **URL** - no key needed.\n3. **Memory** - recalls your watchlist and preferences across sessions (no key).\n\n## How it works\n1. **Chat Input β†’ Agent** - the product you're shopping for.\n2. The Agent searches the web (and News for coupons), reads a listing with URL, computes savings with the Calculator, and checks Memory for anything you tracked before.\n3. **Agent β†’ Chat Output** - the best deals as a structured table (retailer, price, link, notes).\n\n## Run\nOpen **Playground**, name a product, and run.", - "template_notes.research_translation_loop.60bb882f": "# πŸ“– README\nThis template translates research paper summaries on ArXiv into Portuguese and summarizes them. \n Using **OpenXFlow’s looping mechanism**, the template iterates through multiple research papers, translates them with the **OpenAI** model component, and outputs an aggregated version of all translated papers. \n\n## Quick start\n1. Configure your **Model Provider** with your API credentials. \n2. In the **Playground**, enter a query related to a research topic (for example, β€œQuantum Computing Advancements”). \n\n The flow fetches a list of research papers from ArXiv matching the query. Each paper in the retrieved list is processed one-by-one using the OpenXFlow **Loop component**. \n\n The abstract of each paper is translated into Portuguese by the **OpenAI** model component. \n\n Once all papers are translated, the system aggregates them into a **single structured output**.", + "template_notes.research_translation_loop.c0723e26": "# πŸ“– README\nThis template translates research paper summaries on ArXiv into Portuguese and summarizes them. \n Using **OpenXFlow’s looping mechanism**, the template iterates through multiple research papers, translates them with the **OpenAI** model component, and outputs an aggregated version of all translated papers. \n\n## Quick start\n1. Configure your **Model Provider** with your API credentials. \n2. In the **Playground**, enter a query related to a research topic (for example, β€œQuantum Computing Advancements”). \n\n The flow fetches a list of research papers from ArXiv matching the query. Each paper in the retrieved list is processed one-by-one using the OpenXFlow **Loop component**. \n\n The abstract of each paper is translated into Portuguese by the **OpenAI** model component. \n\n Once all papers are translated, the system aggregates them into a **single structured output**.", "template_notes.saas_pricing.d9213061": "# πŸ’΅ SaaS Pricing\n\nCalculate the right monthly subscription price for a software product β€” from costs, profit margin, and subscriber count. The Agent can also pull the numbers from your own database.\n\n## Setup\n1. **Agent** β€” add your LLM key. ⚠️ **Use a capable model** (e.g. a flagship GPT / Claude / Gemini Pro tier). The Agent calls the **SQL Database** tool and does multi-step math, so smaller/faster models may loop and hit the recursion limit when querying the DB.\n2. **SQL Database** *(optional)* β€” set your **Database URL** to let the Agent query real cost / subscriber data from your own database. Leave it unset to just type the numbers in the chat.\n3. The Agent has a built-in **calculator** and **current-date** tool.\n\n## How it works\n1. **Chat Input β†’ Agent** β€” your request (e.g. \"subscription price for Beta CRM using my database\", or \"price: infra 2000, support 1000, dev 3000, margin 30%, 200 subscribers\").\n2. The Agent queries the **SQL Database** when connected, then computes the minimum price per subscriber with its built-in calculator.\n3. **Agent β†’ Chat Output** β€” the recommended pricing breakdown.\n\n## Run\nOpen the **Playground**, describe the product (or name one in your connected database), and run.\n\n> Tip: when using the database, mention the table/product so the Agent queries it in one shot β€” and prefer a capable model to avoid tool-call loops.", "template_notes.seo_keyword_generator.11d10888": "# πŸ” SEO Keyword Generator\n\nGenerate SEO keywords and phrases for a topic with an **Agent**.\n\n## Setup\n1. **Agent** - add your LLM key (any provider in the Model dropdown).\n2. **Prompt Template** - set the topic / niche and tune the keyword instructions.\n\n## How it works\n1. **Prompt (instructions) β†’ Agent (System Prompt)** - steers how keywords are generated.\n2. **Prompt (topic) β†’ Agent (Input)** - the subject to generate keywords for.\n3. **Agent β†’ Chat Output** - the SEO keyword list.\n\n## Run\nOpen **Playground**, set your topic in the Prompt, and run.", "template_notes.sequential_tasks_agents.4b51b574": "# πŸ”— Sequential Tasks Agents\n\nThree agents working in sequence - each does one task and passes its result to the next.\n\n## Setup\nAdd your LLM key on each **Agent**. The **Web Search** tool needs no key.\n\n## How it works\n1. **Research Agent** (Web Search) - gathers information on your topic.\n2. **Analysis Agent** - analyzes the findings and draws conclusions.\n3. **Writer Agent** - synthesizes research + analysis into a final report.\n\n## Run\nType a topic or question in the chat and run; the final report appears in Chat Output.", @@ -3700,14 +3704,5 @@ "template_notes.text_sentiment_analysis.87f5729f": "# πŸ“Š Text Sentiment Analysis\n\nUpload a text file (a review, an email, feedback, an article) and the flow analyzes its sentiment **two complementary ways** β€” a quick direct classification and a deeper summarize-then-analyze pass. Each result comes back as a clean JSON object.\n\n## Setup\n1. **File** β€” upload the text to analyze (PDF, TXT, MD, DOCX, …).\n2. **Agents** β€” add your LLM key on each of the three Agents (Model dropdown). No other keys needed.\n\n## What you get (two outputs)\n**1) Direct classification** β€” File β†’ Agent β†’ Chat Output\nReads the text in one pass and returns:\n- `message_category` β€” e.g. *Positive Feedback*, *Complaint*, *Question*\n- `message_sentiment` β€” *Positive* / *Negative* / *Neutral*\n\n**2) Summary & insights** β€” File β†’ Agent (summarize) β†’ Agent (analyze) β†’ Chat Output\nFirst condenses the text, then analyzes the summary, returning:\n- `summary` β€” a short recap of the text\n- `action_item` β€” a concrete next step suggested by the content\n- `sentiment` β€” overall sentiment of the summarized view\n\nThe two can differ on nuanced text β€” a glowing review with a few complaints may read *Positive* directly but *Neutral* once distilled. Comparing them is the point.\n\n## Run\nUpload a file, open the **Playground**, and run. Swap the file to analyze different texts.\n\n## Tips\n- Clear, well-formatted text gives better results.\n- Each Agent's behavior is set in its **Agent Instructions** / the **Prompt** feeding it β€” edit those to change the categories, fields, or tone.", "template_notes.travel_planning_agents.fcc662a4": "# ✈️ Travel Planning Agents\n\nA team of three **Agents** plans a trip end to end: one researches the destination on the web, one drafts the itinerary, and one works out the budget.\n\n## Setup\nAdd your LLM key on each of the three **Agents** (Model dropdown). **Web Search**, **URL** and **Calculator** tools need no key.\n\n## How it works\n1. **Chat Input β†’ Research Agent** (+ **Web Search**) - gathers current info about the destination.\n2. **β†’ Itinerary Agent** (+ **URL**) - reads relevant pages and drafts the plan.\n3. **β†’ Budget Agent** (+ **Calculator**) - estimates costs and finalizes the trip.\n4. **β†’ Chat Output** - the complete travel plan.\n\n## Run\nOpen **Playground**, describe your trip (destination, dates, budget, interests), and run.", "template_notes.twitter_thread_generator.7e45a5a9": "# 🐦 Twitter Thread Generator\n\nTurn a project brief into an engaging multi-tweet **thread**. Rebuilt on the new **Agent** core component.\n\n## Setup\n1. **Agent** - add your LLM key (any provider in the Model dropdown).\n2. **Prompt Template** - tune the thread rules: PROFILE_TYPE, PROFILE_DETAILS, CONTENT_GUIDELINES, TONE_AND_STYLE, OUTPUT_FORMAT, OUTPUT_LANGUAGE.\n\n## How it works\n1. **Prompt β†’ Agent (System Prompt)** - the instructions + style guidelines steer the Agent.\n2. **Chat Input β†’ Agent (Input)** - your project context / what the thread is about.\n3. **Agent β†’ Chat Output** - the finished tweet or numbered thread.\n\n## Run\nOpen **Playground**, type (or keep) the context in Chat Input, and run.", - "template_notes.vector_store_rag.cdd6e091": "# Vector Store RAG\n\nAsk questions over a **Knowledge Base** (retrieval-augmented generation), fully **bundle-free** - the core **Knowledge** component, no external vector database. Rebuilt on the new **Agent** core component (replaces the Language Model).\n\n## Setup\n1. Point the **Knowledge** node at a populated Knowledge Base - pick one in the dropdown, or ingest documents first (e.g. via the Document Q&A flow).\n2. **Agent** - add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input β†’ Knowledge (Retrieve)** - semantic search for the most relevant chunks.\n2. **Parser β†’ Prompt** - formats the chunks as context alongside your question.\n3. **Agent β†’ Chat Output** - answers using only that retrieved context.", - "components.altkagent.description.4da568f4": "ALTK was retired in OpenXFlow 1.13.0. Replace this node with the Agent component.", - "components.knowledge.description.87015105": "Ingest into or retrieve from an OpenXFlow knowledge base.", - "components.knowledgeingestion.description.4248eab8": "Create or update knowledge in OpenXFlow.", - "components.memory.description.5e6e6979": "Stores or retrieves stored chat messages from OpenXFlow tables or an external memory.", - "components.memory.inputs.memory.info.8d537dfe": "Retrieve messages from an external memory. If empty, it will use the OpenXFlow tables.", - "components.slackonmessagetrigger.inputs.connection.info.2be79dca": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", - "components.slackonreactiontrigger.inputs.connection.info.2be79dca": "Either a Slack app installation (hosted, or self-managed with a public URL: events arrive through the Slack Events API) or an app-level token connection (OpenXFlow Desktop, or an instance Slack cannot reach: events arrive over Socket Mode). Required before the trigger can be armed.", - "components.storemessage.description.ab58b04c": "Stores a chat message or text into OpenXFlow tables or an external memory.", - "components.storemessage.inputs.memory.info.bdde8480": "The external memory to store the message. If empty, it will use the OpenXFlow tables." + "template_notes.vector_store_rag.cdd6e091": "# Vector Store RAG\n\nAsk questions over a **Knowledge Base** (retrieval-augmented generation), fully **bundle-free** - the core **Knowledge** component, no external vector database. Rebuilt on the new **Agent** core component (replaces the Language Model).\n\n## Setup\n1. Point the **Knowledge** node at a populated Knowledge Base - pick one in the dropdown, or ingest documents first (e.g. via the Document Q&A flow).\n2. **Agent** - add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input β†’ Knowledge (Retrieve)** - semantic search for the most relevant chunks.\n2. **Parser β†’ Prompt** - formats the chunks as context alongside your question.\n3. **Agent β†’ Chat Output** - answers using only that retrieved context." } From 20ef058a7e0231cd227ddaff24e363262e384bf7 Mon Sep 17 00:00:00 2001 From: lien Date: Wed, 21 Oct 2026 00:24:43 +0800 Subject: [PATCH 02/32] =?UTF-8?q?fix(telemetry):=20=E4=BF=AE=E5=A4=8D=20Wi?= =?UTF-8?q?ndows=20=E8=BF=9B=E7=A8=8B=E6=8E=A2=E6=B5=8B=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E7=9A=84=E5=86=99=E5=85=A5=E5=99=A8=E5=90=AF=E5=8A=A8=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E5=92=8C=E8=BF=9B=E7=A8=8B=E8=AF=AF=E7=BB=88=E6=AD=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/telemetry_writer/service.py | 17 ++++----- .../telemetry_writer/test_process_liveness.py | 38 +++++++++++++++++++ 2 files changed, 45 insertions(+), 10 deletions(-) create mode 100644 src/backend/tests/unit/services/telemetry_writer/test_process_liveness.py diff --git a/src/backend/base/langflow/services/telemetry_writer/service.py b/src/backend/base/langflow/services/telemetry_writer/service.py index 1461e25100..4a61babe1f 100644 --- a/src/backend/base/langflow/services/telemetry_writer/service.py +++ b/src/backend/base/langflow/services/telemetry_writer/service.py @@ -37,6 +37,7 @@ from typing import TYPE_CHECKING, Any from uuid import UUID +import psutil from lfx.log.logger import logger from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine @@ -212,16 +213,12 @@ def drain(self) -> list[dict[str, Any]]: def _pid_alive(pid: int) -> bool: - """Return True if ``pid`` corresponds to a live process.""" - if pid <= 0: - return False - try: - os.kill(pid, 0) - except ProcessLookupError: - return False - except PermissionError: - return True - return True + """Check process existence without signaling or terminating the worker. + + Unlike POSIX, Windows treats ``os.kill(pid, 0)`` as termination and raises + WinError 87 for missing PIDs. psutil handles these platform differences. + """ + return pid > 0 and psutil.pid_exists(pid) class TelemetryWriterService(Service): diff --git a/src/backend/tests/unit/services/telemetry_writer/test_process_liveness.py b/src/backend/tests/unit/services/telemetry_writer/test_process_liveness.py new file mode 100644 index 0000000000..079bd0d3ca --- /dev/null +++ b/src/backend/tests/unit/services/telemetry_writer/test_process_liveness.py @@ -0,0 +1,38 @@ +"""Process probes must recognize dead workers without signaling live workers.""" + +import subprocess +import sys + +import pytest +from langflow.services.telemetry_writer.service import _pid_alive + + +@pytest.mark.parametrize("pid", [-1, 0]) +def test_pid_alive_rejects_nonpositive_pids(pid: int) -> None: + assert _pid_alive(pid) is False + + +def test_pid_alive_recognizes_exited_process() -> None: + with subprocess.Popen([sys.executable, "-c", "pass"]) as process: # noqa: S603 + process.wait(timeout=10) + assert _pid_alive(process.pid) is False + + +def test_pid_alive_leaves_running_process_responsive() -> None: + # Only probe a disposable child: os.kill(pid, 0) terminates Windows + # processes instead of performing the harmless existence check on POSIX. + with subprocess.Popen( # noqa: S603 + [sys.executable, "-c", "import sys; print(sys.stdin.readline().strip(), flush=True)"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) as process: + try: + assert _pid_alive(process.pid) is True + stdout, _ = process.communicate("still-alive\n", timeout=10) + assert process.returncode == 0 + assert stdout.strip() == "still-alive" + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=10) From a81550b0ff2f3afb7633236d8dcaa581e9e04fb1 Mon Sep 17 00:00:00 2001 From: lien Date: Wed, 21 Oct 2026 06:10:15 +0800 Subject: [PATCH 03/32] fix(ci): repair PR checks and add OCI preview packaging --- .github/workflows/ci.yml | 2 +- .../openxflow-production-release.yml | 112 +++++++++++++++++- scripts/ci/test_ci_workflow.py | 27 +++++ scripts/ci/test_openxflow_release_workflow.py | 18 ++- src/lfx/src/lfx/_assets/component_index.json | 18 +-- 5 files changed, 161 insertions(+), 16 deletions(-) create mode 100644 scripts/ci/test_ci_workflow.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc8c217192..834145d19a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -394,7 +394,7 @@ jobs: - name: Checkout code uses: actions/checkout@v7 with: - ref: ${{ inputs.ref || github.ref }} + ref: ${{ inputs.ref || github.event.pull_request.head.sha || github.sha }} # SeaweedFS's S3 gateway: MinIO no longer publishes pullable images. A service # container cannot pass it the "server" command, so start it here. With no S3 diff --git a/.github/workflows/openxflow-production-release.yml b/.github/workflows/openxflow-production-release.yml index bf95449819..076997ec19 100644 --- a/.github/workflows/openxflow-production-release.yml +++ b/.github/workflows/openxflow-production-release.yml @@ -2,7 +2,7 @@ name: OpenXFlow Production Release on: pull_request: - types: [closed] + types: [opened, synchronize, reopened, closed] branches: [main] permissions: @@ -13,13 +13,119 @@ env: IMAGE_NAME: lien0219/openxflow PRODUCTION_REPOSITORY: lien0219/openxflow -# Every merged PR gets its own run. Only the floating main promotion is -# serialized, so a slow build cannot discard an immutable source image. +# PR updates produce a verified artifact without registry publication. Every +# merged PR gets its own release run; only floating main promotion is serialized. concurrency: group: openxflow-production-${{ github.event.pull_request.merge_commit_sha || github.run_id }} cancel-in-progress: false jobs: + preview-package: + name: Build and verify PR package + if: >- + github.repository == 'lien0219/openxflow' && + github.event_name == 'pull_request' && + github.event.action != 'closed' && + github.event.pull_request.base.ref == 'main' && + github.event.pull_request.base.repo.full_name == 'lien0219/openxflow' + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - name: Check out PR head + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 1 + persist-credentials: false + + - name: Set up emulation and Buildx + uses: docker/setup-qemu-action@v3 + - uses: docker/setup-buildx-action@v3 + + - name: Materialize the PR image verification plan + id: preview-plan + shell: bash + run: | + set -Eeuo pipefail + source="$(git rev-parse HEAD)" + version="$(cat deploy/IMAGE_VERSION)" + created="$(git show -s --format=%cI "$source")" + python - "$source" "$version" "$created" > /tmp/openxflow-pr-plan.json <<'PY' + import json + import sys + print(json.dumps({"source": sys.argv[1], "version": sys.argv[2], "created": sys.argv[3]})) + PY + { + echo "source=$source" + echo "version=$version" + echo "created=$created" + } >> "$GITHUB_OUTPUT" + + - name: Build the PR OCI package + uses: docker/build-push-action@v6 + with: + context: . + file: ./docker/build_and_push.Dockerfile + target: full + platforms: linux/amd64,linux/arm64 + push: false + outputs: type=oci,dest=/tmp/openxflow-pr.oci.tar + build-args: | + OPENXFLOW_VERSION=${{ steps.preview-plan.outputs.version }} + OPENXFLOW_REVISION=${{ steps.preview-plan.outputs.source }} + OPENXFLOW_CREATED=${{ steps.preview-plan.outputs.created }} + OPENXFLOW_SOURCE=https://github.com/lien0219/openxflow + OPENXFLOW_URL=https://github.com/lien0219/openxflow + labels: | + org.opencontainers.image.title=OpenXFlow + org.opencontainers.image.description=OpenXFlow workflow automation platform + org.opencontainers.image.version=${{ steps.preview-plan.outputs.version }} + org.opencontainers.image.revision=${{ steps.preview-plan.outputs.source }} + org.opencontainers.image.created=${{ steps.preview-plan.outputs.created }} + org.opencontainers.image.source=https://github.com/lien0219/openxflow + org.opencontainers.image.url=https://github.com/lien0219/openxflow + cache-from: type=gha,scope=openxflow-pr-${{ github.event.pull_request.number }} + cache-to: type=gha,mode=max,scope=openxflow-pr-${{ github.event.pull_request.number }} + provenance: mode=max + sbom: true + + - name: Scan the unpushed PR OCI package + uses: aquasecurity/trivy-action@0.28.0 + with: + input: /tmp/openxflow-pr.oci.tar + format: table + exit-code: "1" + ignore-unfixed: "false" + severity: CRITICAL,HIGH + scanners: vuln,secret + + - name: Verify PR package attestations and runtime probes + shell: bash + run: | + set -Eeuo pipefail + python scripts/ci/openxflow_verify_image.py \ + --archive /tmp/openxflow-pr.oci.tar \ + --plan /tmp/openxflow-pr-plan.json \ + --output-dir /tmp/openxflow-pr-image-verification + + - name: Retain PR image verification evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: openxflow-pr-image-verification-${{ github.event.pull_request.number }}-${{ steps.preview-plan.outputs.source }} + path: /tmp/openxflow-pr-image-verification + if-no-files-found: ignore + retention-days: 7 + + - name: Upload verified PR OCI package + uses: actions/upload-artifact@v4 + with: + name: openxflow-pr-oci-${{ github.event.pull_request.number }}-${{ steps.preview-plan.outputs.source }} + path: /tmp/openxflow-pr.oci.tar + if-no-files-found: error + retention-days: 7 + validate-merge: name: Validate exact merged main source if: >- diff --git a/scripts/ci/test_ci_workflow.py b/scripts/ci/test_ci_workflow.py new file mode 100644 index 0000000000..f54460b250 --- /dev/null +++ b/scripts/ci/test_ci_workflow.py @@ -0,0 +1,27 @@ +"""Regression checks for CI checkout refs that differ between push and PR events.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parents[2] + + +def _load(path: str) -> dict: + class Loader(yaml.SafeLoader): + pass + + # PyYAML 1.1 treats the YAML 1.2 ``on`` key as a boolean. + Loader.add_constructor("tag:yaml.org,2002:bool", lambda loader, node: loader.construct_scalar(node)) + return yaml.load((ROOT / path).read_text(encoding="utf-8"), Loader=Loader) # noqa: S506 + + +def test_object_storage_uses_a_stable_pr_checkout_ref() -> None: + workflow = _load(".github/workflows/ci.yml") + checkout = next( + step for step in workflow["jobs"]["test-object-storage"]["steps"] if step.get("name") == "Checkout code" + ) + + assert checkout["with"]["ref"] == "${{ inputs.ref || github.event.pull_request.head.sha || github.sha }}" diff --git a/scripts/ci/test_openxflow_release_workflow.py b/scripts/ci/test_openxflow_release_workflow.py index 21223a9121..90b232a2bf 100644 --- a/scripts/ci/test_openxflow_release_workflow.py +++ b/scripts/ci/test_openxflow_release_workflow.py @@ -19,10 +19,13 @@ class Loader(yaml.SafeLoader): return yaml.load(WORKFLOW.read_text(encoding="utf-8"), Loader=Loader) -def test_only_closed_merged_main_pr_can_start_workflow(): +def test_workflow_builds_pr_artifacts_and_publishes_only_after_merge(): workflow = load() assert set(workflow["on"]) == {"pull_request"} - assert workflow["on"]["pull_request"] == {"types": ["closed"], "branches": ["main"]} + assert workflow["on"]["pull_request"] == { + "types": ["opened", "synchronize", "reopened", "closed"], + "branches": ["main"], + } admission = workflow["jobs"]["validate-merge"]["if"] for expression in [ "github.repository == 'lien0219/openxflow'", @@ -33,6 +36,14 @@ def test_only_closed_merged_main_pr_can_start_workflow(): "github.event.pull_request.merge_commit_sha != ''", ]: assert expression in admission + preview = workflow["jobs"]["preview-package"] + assert "github.event.action != 'closed'" in preview["if"] + checkout = next(step for step in preview["steps"] if step.get("name") == "Check out PR head") + assert checkout["with"]["ref"] == "${{ github.event.pull_request.head.sha }}" + build = next(step for step in preview["steps"] if step.get("name") == "Build the PR OCI package") + assert build["with"]["push"] == "false" + assert build["with"]["outputs"] == "type=oci,dest=/tmp/openxflow-pr.oci.tar" + assert any(step.get("uses", "").startswith("actions/upload-artifact@") for step in preview["steps"]) for forbidden in ["workflow_dispatch", "workflow_call", "workflow_run", "schedule", "pull_request_target"]: assert forbidden not in workflow["on"] @@ -70,8 +81,9 @@ def test_retry_runtime_probes_use_the_verified_reuse_coordinate(): def test_workflow_has_no_untrusted_code_trigger_or_manual_sha_input(): + workflow = load() source = WORKFLOW.read_text(encoding="utf-8") assert "pull_request_target" not in source assert "workflow_dispatch" not in source assert "inputs:" not in source - assert "github.event.pull_request.head.sha" not in source + assert "github.event.pull_request.head.sha" not in workflow["jobs"]["validate-merge"]["if"] diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index 2a79128e25..26b6e554d9 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -4448,7 +4448,7 @@ "icon": "layout-template", "legacy": false, "metadata": { - "code_hash": "a6429c7f9670", + "code_hash": "8052088025a1", "component_class_name": "URLComponent", "dependencies": { "dependencies": [ @@ -4460,13 +4460,13 @@ "name": "bs4", "version": "4.15.0" }, - { - "name": "markitdown", - "version": "0.1.6" - }, { "name": "lfx", "version": null + }, + { + "name": "markitdown", + "version": "0.1.6" } ], "total_dependencies": 4 @@ -4567,7 +4567,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import importlib.util\nimport io\nimport os\nimport re\nimport zlib\nfrom urllib.parse import urljoin, urlparse\n\nimport httpx\nfrom bs4 import BeautifulSoup\nfrom markitdown import MarkItDown\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.field_typing.range_spec import RangeSpec\nfrom lfx.helpers.data import safe_convert\nfrom lfx.io import BoolInput, DropdownInput, IntInput, MessageTextInput, Output, SliderInput, TableInput\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\nfrom lfx.utils.request_utils import get_user_agent\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, is_ssrf_protection_enabled, validate_and_resolve_url\nfrom lfx.utils.ssrf_transport import create_ssrf_protected_client, pin_host_for_url\n\n# Constants\nDEFAULT_TIMEOUT = 30\nDEFAULT_MAX_DEPTH = 1\nDEFAULT_FORMAT = \"Text\"\n\n# HTTP status codes that carry a redirect Location header (RFC 9110).\nREDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308})\n\n# Maximum number of redirects to follow. Each hop is re-validated for SSRF safety;\n# matches httpx's default and the API Request component.\nMAX_REDIRECTS = 20\n\n# Default ports per scheme, used to compare redirect origins.\nDEFAULT_SCHEME_PORTS = {\"http\": 80, \"https\": 443}\n\n# Memory bounds for untrusted responses. Bodies are streamed and abandoned once the\n# per-response cap or the budget shared by every page of one fetch (all URLs and crawled\n# links) is exceeded, so a hostile server cannot exhaust memory with a huge or endless body.\n# Configurable via LANGFLOW_URL_COMPONENT_MAX_RESPONSE_BYTES / LANGFLOW_URL_COMPONENT_MAX_TOTAL_BYTES\n# (see RuntimeSettings); these are only the fallback used when the settings service isn't available.\nFALLBACK_MAX_RESPONSE_BYTES = 10 * 1024 * 1024\nFALLBACK_MAX_TOTAL_BYTES = 100 * 1024 * 1024\n\n# Only advertise codings we can decode with a strict output limit. Raw response bytes\n# are also capped before decoding, including compressed headers and trailing data.\nACCEPT_ENCODING = \"gzip, deflate\"\nALLOWED_CONTENT_ENCODINGS = frozenset({\"gzip\", \"deflate\"})\nZLIB_HEADER_BYTES = 2\n\n\nURL_REGEX = re.compile(\n r\"^(https?:\\/\\/)?\" r\"(www\\.)?\" r\"([a-zA-Z0-9.-]+)\" r\"(\\.[a-zA-Z]{2,})?\" r\"(:\\d+)?\" r\"(\\/[^\\s]*)?$\",\n re.IGNORECASE,\n)\n\nUSER_AGENT = None\n# Check if langflow is installed using importlib.util.find_spec(name))\nif importlib.util.find_spec(\"langflow\"):\n langflow_installed = True\n USER_AGENT = get_user_agent()\nelse:\n langflow_installed = False\n USER_AGENT = \"lfx\"\n\n\nclass URLComponent(Component):\n \"\"\"A component that loads and parses content from web pages recursively.\n\n This component allows fetching content from one or more URLs, with options to:\n - Control crawl depth\n - Prevent crawling outside the root domain\n - Use async loading for better performance\n - Extract either raw HTML or clean text\n - Configure request headers and timeouts\n \"\"\"\n\n display_name = \"URL\"\n description = \"Fetch content from one or more web pages, following links recursively.\"\n documentation: str = \"https://docs.langflow.org/url\"\n icon = \"layout-template\"\n name = \"URLComponent\"\n # Body bytes the current fetch may still read; reset by ``fetch_url_contents``.\n _bytes_remaining = FALLBACK_MAX_TOTAL_BYTES\n # Per-response and total byte caps for the current fetch; resolved from settings by\n # ``fetch_url_contents``. Kept as fallbacks here so the class works if that never runs.\n _max_response_bytes = FALLBACK_MAX_RESPONSE_BYTES\n _max_total_bytes = FALLBACK_MAX_TOTAL_BYTES\n\n inputs = [\n MessageTextInput(\n name=\"urls\",\n display_name=\"URLs\",\n info=\"Enter one or more URLs to crawl recursively, by clicking the '+' button.\",\n is_list=True,\n tool_mode=True,\n placeholder=\"Enter a URL...\",\n list_add_label=\"Add URL\",\n input_types=[\"Message\"],\n ),\n SliderInput(\n name=\"max_depth\",\n display_name=\"Depth\",\n info=(\n \"Controls how many 'clicks' away from the initial page the crawler will go:\\n\"\n \"- depth 1: only the initial page\\n\"\n \"- depth 2: initial page + all pages linked directly from it\\n\"\n \"- depth 3: initial page + direct links + links found on those direct link pages\\n\"\n \"Note: This is about link traversal, not URL path depth.\"\n ),\n value=DEFAULT_MAX_DEPTH,\n range_spec=RangeSpec(min=1, max=5, step=1),\n required=False,\n min_label=\" \",\n max_label=\" \",\n min_label_icon=\"None\",\n max_label_icon=\"None\",\n # slider_input=True\n ),\n BoolInput(\n name=\"prevent_outside\",\n display_name=\"Prevent Outside\",\n info=(\n \"If enabled, only crawls URLs within the same domain as the root URL. \"\n \"This helps prevent the crawler from going to external websites.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"use_async\",\n display_name=\"Use Async\",\n info=(\n \"If enabled, uses asynchronous loading which can be significantly faster \"\n \"but might use more system resources.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"follow_redirects\",\n display_name=\"Follow Redirects\",\n info=(\n \"If enabled, follows HTTP redirects such as httpβ†’https or www/non-www \"\n \"normalization, which most sites rely on to serve their content. When SSRF \"\n \"protection is enabled (the default), every redirect hop is re-validated against \"\n \"the same blocked-IP denylist and DNS-pinned before it is fetched, so following \"\n \"redirects cannot be used to reach internal resources. Disable to capture the \"\n \"first response as-is without following redirects.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n DropdownInput(\n name=\"format\",\n display_name=\"Output Format\",\n info=(\n \"Output Format. Use 'Text' to extract the text from the HTML, \"\n \"'Markdown' to parse the HTML into Markdown format, or 'HTML' \"\n \"for the raw HTML content.\"\n ),\n options=[\"Text\", \"HTML\", \"Markdown\"],\n value=DEFAULT_FORMAT,\n advanced=True,\n ),\n IntInput(\n name=\"timeout\",\n display_name=\"Timeout\",\n info=\"Timeout for the request in seconds.\",\n value=DEFAULT_TIMEOUT,\n required=False,\n advanced=True,\n ),\n TableInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=\"The headers to send with the request\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Header\",\n \"type\": \"str\",\n \"description\": \"Header name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"type\": \"str\",\n \"description\": \"Header value\",\n },\n ],\n value=[{\"key\": \"User-Agent\", \"value\": USER_AGENT}],\n advanced=True,\n input_types=[\"DataFrame\", \"Table\"],\n ),\n BoolInput(\n name=\"filter_text_html\",\n display_name=\"Filter Text/HTML\",\n info=\"If enabled, filters out text/css content type from the results.\",\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"continue_on_failure\",\n display_name=\"Continue on Failure\",\n info=\"If enabled, continues crawling even if some requests fail.\",\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"check_response_status\",\n display_name=\"Check Response Status\",\n info=\"If enabled, checks the response status of the request.\",\n value=False,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"autoset_encoding\",\n display_name=\"Autoset Encoding\",\n info=\"If enabled, automatically sets the encoding of the request.\",\n value=True,\n required=False,\n advanced=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Extracted Pages\", name=\"page_results\", method=\"fetch_content\"),\n Output(display_name=\"Raw Content\", name=\"raw_results\", method=\"fetch_content_as_message\", tool_mode=False),\n ]\n\n @staticmethod\n def _html_extractor(x: str) -> str:\n \"\"\"Extract raw HTML content.\"\"\"\n return x\n\n @staticmethod\n def _text_extractor(x: str) -> str:\n \"\"\"Extract clean text from HTML.\"\"\"\n return BeautifulSoup(x, \"lxml\").get_text()\n\n @staticmethod\n def _markdown_extractor(x: str) -> str:\n \"\"\"Convert HTML to Markdown format.\"\"\"\n stream = io.BytesIO(x.encode(\"utf-8\"))\n result = MarkItDown(enable_plugins=False).convert_stream(stream)\n return result.markdown\n\n @staticmethod\n def validate_url(url: str) -> bool:\n \"\"\"Validates if the given string matches URL pattern.\n\n Args:\n url: The URL string to validate\n\n Returns:\n bool: True if the URL is valid, False otherwise\n \"\"\"\n return bool(URL_REGEX.match(url))\n\n def ensure_url(self, url: str) -> tuple[str, list[str]]:\n \"\"\"Ensures the given string is a valid URL and returns validated IPs for DNS pinning.\n\n Args:\n url: The URL string to validate and normalize\n\n Returns:\n tuple[str, list[str]]: The normalized URL and list of validated IPs for DNS pinning\n\n Raises:\n ValueError: If the URL is invalid or blocked by SSRF protection\n \"\"\"\n url = url.strip()\n if not url.startswith((\"http://\", \"https://\")):\n url = \"https://\" + url\n\n if not self.validate_url(url):\n msg = f\"Invalid URL: {url}\"\n raise ValueError(msg)\n\n # ============================================================================\n # SSRF Protection with DNS Pinning\n # ============================================================================\n # This prevents DNS rebinding attacks by:\n # 1. Resolving DNS and validating IPs during security check\n # 2. Returning the validated IP addresses for DNS pinning\n # 3. Using a custom HTTP transport that forces use of the pinned IPs\n # 4. Ignoring any new DNS resolutions (prevents rebinding)\n #\n # Without DNS pinning, an attacker could:\n # - First DNS lookup: returns public IP (passes validation)\n # - Second DNS lookup: returns internal IP (bypasses protection)\n # - Attack succeeds: accesses internal services\n #\n # With DNS pinning:\n # - First DNS lookup: returns public IP (passes validation)\n # - IPs are pinned and returned\n # - HTTP request: uses pinned IPs directly (no new DNS lookup)\n # - Attack fails: even if DNS changes, we use the validated IPs\n # ============================================================================\n try:\n _validated_url, validated_ips = validate_and_resolve_url(url)\n except SSRFProtectionError as e:\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n else:\n # Log DNS pinning information for security auditing\n if validated_ips:\n logger.debug(f\"SSRF Protection: Using DNS pinning with {len(validated_ips)} validated IP(s) for {url}\")\n\n return url, validated_ips\n\n def _build_http_client(self, url: str, validated_ips: list[str]) -> httpx.AsyncClient:\n \"\"\"Create an HTTP client with DNS pinning for SSRF protection.\n\n Args:\n url: The request URL whose hostname will be pinned\n validated_ips: IPs validated by validate_and_resolve_url for this URL\n\n Returns:\n httpx.AsyncClient: A client with DNS pinning when SSRF protection is enabled\n \"\"\"\n hostname = pin_host_for_url(url) if is_ssrf_protection_enabled() and validated_ips else None\n if hostname:\n client = create_ssrf_protected_client(hostname=hostname, validated_ips=validated_ips)\n else:\n client = httpx.AsyncClient()\n # A client default, so a user-supplied Accept-Encoding header still takes precedence.\n client.headers[\"Accept-Encoding\"] = ACCEPT_ENCODING\n return client\n\n @staticmethod\n def _headers_for_redirect(headers: dict | None, current_url: str, next_url: str) -> dict | None:\n \"\"\"Drop sensitive headers when a redirect crosses to a different origin.\n\n Mirrors httpx's auto-follow behavior so manually following redirects does not\n leak credentials (Authorization / Cookie) to an origin other than the one the\n caller intended them for. Headers are kept only when the redirect stays on the\n same origin (scheme, host, port) or is a direct https upgrade of the same host\n on default ports - the exact cases where httpx keeps the Authorization header.\n \"\"\"\n if not headers:\n return headers\n current, nxt = urlparse(current_url), urlparse(next_url)\n current_port = current.port or DEFAULT_SCHEME_PORTS.get(current.scheme)\n next_port = nxt.port or DEFAULT_SCHEME_PORTS.get(nxt.scheme)\n same_origin = (current.scheme, current.hostname, current_port) == (nxt.scheme, nxt.hostname, next_port)\n https_upgrade = (\n current.hostname == nxt.hostname\n and current.scheme == \"http\"\n and nxt.scheme == \"https\"\n and current_port == DEFAULT_SCHEME_PORTS[\"http\"]\n and next_port == DEFAULT_SCHEME_PORTS[\"https\"]\n )\n if same_origin or https_upgrade:\n return headers\n sensitive = {\"authorization\", \"proxy-authorization\", \"cookie\"}\n return {k: v for k, v in headers.items() if k.lower() not in sensitive}\n\n @staticmethod\n def _resolve_byte_limit(env_var: str, setting_name: str, fallback: int) -> int:\n \"\"\"Resolve one byte cap from settings, or the environment when no service is available.\n\n The settings service applies Langflow's normal configuration precedence. Reading the\n process environment directly is only a fallback for standalone ``lfx`` execution.\n \"\"\"\n try:\n settings_service = get_settings_service()\n except Exception: # noqa: BLE001 - service registry may not be ready\n settings_service = None\n if settings_service is not None:\n return getattr(settings_service.settings, setting_name, fallback)\n\n env_value = os.getenv(env_var)\n if env_value is not None:\n try:\n value = int(env_value)\n except ValueError:\n logger.warning(f\"Ignoring invalid {env_var}={env_value!r}; using {fallback}\")\n return fallback\n if value > 0:\n return value\n logger.warning(f\"Ignoring non-positive {env_var}={env_value!r}; using {fallback}\")\n return fallback\n\n return fallback\n\n @classmethod\n def _resolve_byte_limits(cls) -> tuple[int, int]:\n \"\"\"Read the configurable per-response and total byte caps for the current fetch.\"\"\"\n return (\n cls._resolve_byte_limit(\n \"LANGFLOW_URL_COMPONENT_MAX_RESPONSE_BYTES\",\n \"url_component_max_response_bytes\",\n FALLBACK_MAX_RESPONSE_BYTES,\n ),\n cls._resolve_byte_limit(\n \"LANGFLOW_URL_COMPONENT_MAX_TOTAL_BYTES\",\n \"url_component_max_total_bytes\",\n FALLBACK_MAX_TOTAL_BYTES,\n ),\n )\n\n async def _read_bounded_text(self, response: httpx.Response) -> str:\n \"\"\"Read a streamed response body without exceeding the per-response or total byte budget.\n\n Encoded and decoded bytes are each bounded. The larger count is charged against the\n total budget even when the body is rejected, so a server that keeps answering with\n oversized bodies cannot make a crawl download without bound. Decoder errors charge\n the reserved output bound because zlib may discard output before raising.\n\n Raises:\n httpx.HTTPError: If the body is (or declares to be) larger than the remaining budget,\n or uses a content coding whose decompression cannot be bounded.\n \"\"\"\n codings = [c.strip().lower() for c in response.headers.get(\"content-encoding\", \"\").split(\",\")]\n codings = [c for c in codings if c not in {\"\", \"identity\"}]\n if len(codings) > 1 or (codings and codings[0] not in ALLOWED_CONTENT_ENCODINGS):\n msg = f\"Unsupported content encoding from {response.url}: {', '.join(codings)}\"\n raise httpx.HTTPError(msg)\n\n limit = min(self._max_response_bytes, self._bytes_remaining)\n too_large = f\"Response from {response.url} exceeds the {limit} byte limit\"\n declared = response.headers.get(\"content-length\", \"\")\n if declared.isdigit() and int(declared) > limit:\n raise httpx.HTTPError(too_large)\n\n body = bytearray()\n raw_bytes = decoded_bytes = 0\n coding = codings[0] if codings else None\n decompressor = (\n zlib.decompressobj(zlib.MAX_WBITS | 16 if coding == \"gzip\" else zlib.MAX_WBITS) if coding else None\n )\n first_decode = True\n deflate_prefix = b\"\"\n try:\n # HTTPX's automatic decoder can allocate unbounded output or retain an\n # endless tail in zlib.unused_data without yielding any decoded bytes.\n async for raw_chunk in response.aiter_raw():\n raw_bytes += len(raw_chunk)\n if raw_bytes > limit:\n raise httpx.HTTPError(too_large)\n chunk = raw_chunk\n if decompressor is not None:\n if decompressor.eof:\n msg = f\"Trailing data after compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n if coding == \"deflate\" and first_decode:\n # HTTP deflate may omit the zlib wrapper. Wait for enough\n # header bytes before trying the same fallback as HTTPX.\n chunk = deflate_prefix + chunk\n if len(chunk) < ZLIB_HEADER_BYTES:\n deflate_prefix = chunk\n continue\n max_output = limit - decoded_bytes + 1\n try:\n try:\n decoded = decompressor.decompress(chunk, max_output)\n except zlib.error:\n if coding != \"deflate\" or not first_decode:\n raise\n decompressor = zlib.decompressobj(-zlib.MAX_WBITS)\n decoded = decompressor.decompress(chunk, max_output)\n except zlib.error as exc:\n # zlib can produce output before detecting a bad checksum,\n # then raise without returning it. Charge the reserved bound.\n decoded_bytes = limit + 1\n msg = f\"Invalid compressed response from {response.url}\"\n raise httpx.DecodingError(msg) from exc\n first_decode = False\n decoded_bytes += len(decoded)\n if decoded_bytes > limit:\n raise httpx.HTTPError(too_large)\n if decompressor.unused_data:\n msg = f\"Trailing data after compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n chunk = decoded\n else:\n decoded_bytes += len(chunk)\n # Check decoded output before retaining it in the response body.\n body.extend(chunk)\n if decompressor is not None and not decompressor.eof:\n msg = f\"Incomplete compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n finally:\n # Charge encoded overhead as well as expanded output, including rejected\n # responses. Neither representation may bypass the shared fetch budget.\n self._bytes_remaining -= max(raw_bytes, decoded_bytes)\n # Same character decoding as ``response.text``. A complete zlib stream has\n # already emitted all its output, so no unbounded decoder flush is needed.\n return body.decode(response.encoding or \"utf-8\", errors=\"replace\")\n\n async def _process_response(self, response: httpx.Response) -> tuple[str, dict]:\n \"\"\"Turn a final (non-redirect) streamed response into its HTML content and metadata.\n\n Args:\n response: The HTTP response to process, opened with ``stream=True``\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n if self.check_response_status:\n response.raise_for_status()\n\n content_type = response.headers.get(\"content-type\", \"\").lower()\n\n # Filter out CSS files if requested\n if self.filter_text_html and \"text/css\" in content_type:\n logger.debug(f\"Skipping CSS file: {response.url}\")\n return \"\", {}\n\n # Get the HTML content\n html_content = await self._read_bounded_text(response)\n\n # Extract metadata\n metadata = {\n \"source\": str(response.url),\n \"title\": \"\",\n \"description\": \"\",\n \"content_type\": content_type,\n \"language\": \"\",\n }\n\n # Try to extract title and description from HTML\n try:\n soup = BeautifulSoup(html_content, \"lxml\")\n except Exception: # noqa: BLE001\n # Broad exception is acceptable here - metadata extraction is optional\n # and we don't want to fail the entire request if it fails\n logger.debug(f\"Failed to extract metadata from {response.url}\")\n else:\n if soup.title:\n metadata[\"title\"] = soup.title.string or \"\"\n\n # Try to get description from meta tags\n meta_desc = soup.find(\"meta\", attrs={\"name\": \"description\"})\n if meta_desc and meta_desc.get(\"content\"):\n metadata[\"description\"] = meta_desc[\"content\"]\n\n # Try to get language\n html_tag = soup.find(\"html\")\n if html_tag and html_tag.get(\"lang\"):\n metadata[\"language\"] = html_tag[\"lang\"]\n\n return html_content, metadata\n\n async def _fetch_with_revalidated_redirects(\n self, url: str, validated_ips: list[str], headers: dict\n ) -> tuple[str, dict]:\n \"\"\"Fetch ``url``, following redirects manually and re-validating every hop.\n\n This closes an SSRF bypass: with redirects enabled, httpx would otherwise\n auto-follow a redirect from a validated public URL to an internal address that\n the pinned transport never checked (the redirect target is a different host, so\n it is not in the pin map). Each redirect ``Location`` is resolved (relative\n locations included) and re-validated with ``ensure_url`` - the same private/\n loopback/link-local denylist and DNS pinning applied to the initial request -\n before any connection to it is made. A blocked hop raises ``ValueError`` (unless\n ``continue_on_failure``); the chain is capped at ``MAX_REDIRECTS`` hops.\n\n Args:\n url: The URL to fetch\n validated_ips: Validated IPs for DNS pinning of the initial URL\n headers: HTTP headers to send\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n current_url = url\n current_ips = validated_ips\n\n for _ in range(MAX_REDIRECTS + 1):\n # Stream so redirect bodies are never read and the final body is read bounded.\n async with (\n self._build_http_client(current_url, current_ips) as client,\n client.stream(\n \"GET\", current_url, headers=headers, timeout=self.timeout, follow_redirects=False\n ) as response,\n ):\n location = response.headers.get(\"location\")\n if response.status_code not in REDIRECT_STATUS_CODES or not location:\n # Not a redirect (or no Location header) - this is the final response.\n return await self._process_response(response)\n\n # Resolve relative redirects against the current URL.\n next_url = urljoin(current_url, location)\n\n # Re-validate the redirect target with the same SSRF denylist + DNS pinning.\n try:\n validated_next_url, current_ips = self.ensure_url(next_url)\n except (ValueError, SSRFProtectionError) as e:\n if self.continue_on_failure:\n logger.warning(f\"Skipping blocked or invalid redirect to {next_url}: {e}\")\n return \"\", {}\n msg = f\"SSRF Protection: blocked redirect to {next_url}: {e}\"\n raise ValueError(msg) from e\n\n headers = self._headers_for_redirect(headers, current_url, next_url)\n current_url = validated_next_url\n\n # Exhausted the redirect budget.\n if self.continue_on_failure:\n logger.warning(f\"Exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}\")\n return \"\", {}\n msg = f\"SSRF Protection: exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}\"\n raise ValueError(msg)\n\n async def _fetch_url_with_pinning(self, url: str, validated_ips: list[str], headers: dict) -> tuple[str, dict]:\n \"\"\"Fetch a single URL with DNS pinning protection, following redirects safely.\n\n Args:\n url: The URL to fetch\n validated_ips: Validated IPs for DNS pinning\n headers: HTTP headers to send\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n try:\n # When SSRF protection is enabled we must follow redirects manually so each hop\n # is re-validated and DNS-pinned; letting httpx auto-follow would connect to the\n # redirect target without pinning (different host, not in the pin map) and re-open\n # the SSRF hole that DNS pinning closes. With protection disabled there is no pin\n # to bypass, so redirects follow the requests httpx itself builds for each hop.\n if self.follow_redirects and is_ssrf_protection_enabled():\n return await self._fetch_with_revalidated_redirects(url, validated_ips, headers)\n\n async with self._build_http_client(url, validated_ips) as client:\n request = client.build_request(\"GET\", url, headers=headers, timeout=self.timeout)\n # httpx auto-follow reads every redirect body into memory without a limit, so hop\n # manually via ``next_request`` (same cookies, headers and redirect cap as httpx).\n for _ in range(MAX_REDIRECTS + 1):\n response = await client.send(request, stream=True, follow_redirects=False)\n try:\n if not (self.follow_redirects and response.next_request):\n return await self._process_response(response)\n request = response.next_request\n finally:\n await response.aclose()\n msg = \"Exceeded maximum allowed redirects.\"\n raise httpx.TooManyRedirects(msg, request=request)\n\n except httpx.HTTPError as e:\n if self.continue_on_failure:\n logger.warning(f\"Failed to fetch {url}: {e}\")\n return \"\", {}\n raise\n\n async def _crawl_recursive(\n self, start_url: str, validated_ips: list[str], headers: dict, visited: set, depth: int = 0\n ) -> list[dict]:\n \"\"\"Recursively crawl URLs with DNS pinning protection.\n\n Args:\n start_url: The URL to start crawling from\n validated_ips: Validated IPs for DNS pinning\n headers: HTTP headers to send\n visited: Set of already visited URLs\n depth: Current crawl depth\n\n Returns:\n list[dict]: List of documents with content and metadata\n \"\"\"\n if depth >= self.max_depth or start_url in visited:\n return []\n if self._bytes_remaining <= 0:\n logger.warning(f\"Skipping {start_url}: reached the {self._max_total_bytes} byte limit for this fetch\")\n return []\n\n visited.add(start_url)\n documents = []\n\n # Fetch the current URL\n html_content, metadata = await self._fetch_url_with_pinning(start_url, validated_ips, headers)\n\n # Redirects may have landed on a different canonical URL (http->https, www\n # normalization). Resolve relative links and the prevent_outside check against\n # the URL the content actually came from, and mark it visited so links back to\n # the canonical form are not re-crawled.\n base_url = metadata.get(\"source\") or start_url\n visited.add(base_url)\n\n if not html_content:\n return documents\n\n # Extract content based on format\n extractors = {\n \"HTML\": self._html_extractor,\n \"Markdown\": self._markdown_extractor,\n \"Text\": self._text_extractor,\n }\n extractor = extractors.get(self.format, self._text_extractor)\n extracted_content = extractor(html_content)\n\n # Add the document\n documents.append(\n {\n \"page_content\": extracted_content,\n \"metadata\": metadata,\n }\n )\n\n # If we haven't reached max depth, extract and follow links\n if depth < self.max_depth - 1:\n try:\n soup = BeautifulSoup(html_content, \"lxml\")\n links = soup.find_all(\"a\", href=True)\n\n for link in links:\n if self._bytes_remaining <= 0:\n break\n href = link[\"href\"]\n # Resolve relative URLs\n absolute_url = urljoin(base_url, href)\n\n # Skip if already visited\n if absolute_url in visited:\n continue\n\n # Check if we should prevent going outside the domain\n if self.prevent_outside:\n start_domain = urlparse(base_url).netloc\n link_domain = urlparse(absolute_url).netloc\n if start_domain != link_domain:\n continue\n\n # Validate and crawl the linked URL\n try:\n _, link_validated_ips = self.ensure_url(absolute_url)\n sub_docs = await self._crawl_recursive(\n absolute_url, link_validated_ips, headers, visited, depth + 1\n )\n documents.extend(sub_docs)\n except (ValueError, SSRFProtectionError) as e:\n if self.continue_on_failure:\n logger.warning(f\"Skipping {absolute_url}: {e}\")\n continue\n raise\n\n except Exception: # noqa: BLE001\n # Broad exception is acceptable here - link extraction is optional\n # and we don't want to fail the entire crawl if one page has issues\n logger.debug(f\"Failed to extract links from {start_url}\")\n\n return documents\n\n async def fetch_url_contents(self) -> list[dict]:\n \"\"\"Load documents from the configured URLs with SSRF protection and DNS pinning.\n\n This method implements comprehensive SSRF (Server-Side Request Forgery) protection\n using DNS pinning to prevent DNS rebinding attacks. Each URL is validated and its\n DNS resolution is pinned before any HTTP requests are made.\n\n Returns:\n list[dict]: List of documents with content and metadata\n\n Raises:\n ValueError: If no valid URLs are provided or if there's an error loading documents\n \"\"\"\n self._max_response_bytes, self._max_total_bytes = self._resolve_byte_limits()\n self._bytes_remaining = self._max_total_bytes\n try:\n # Validate all URLs and get their validated IPs for DNS pinning\n validated_urls = []\n first_validation_error: Exception | None = None\n for url in self.urls:\n if not url.strip():\n continue\n try:\n normalized_url, validated_ips = self.ensure_url(url)\n validated_urls.append((normalized_url, validated_ips))\n except (ValueError, SSRFProtectionError) as e:\n # Remember the first rejection so its real cause (for example an\n # SSRF block) can be surfaced if no URL survives validation,\n # instead of being hidden behind \"No valid URLs provided\".\n if first_validation_error is None:\n first_validation_error = e\n if self.continue_on_failure:\n logger.warning(f\"Skipping invalid URL {url}: {e}\")\n continue\n raise\n\n # Remove duplicates while preserving order\n seen = set()\n unique_validated_urls = []\n for url, ips in validated_urls:\n if url not in seen:\n seen.add(url)\n unique_validated_urls.append((url, ips))\n\n logger.debug(f\"Validated {len(unique_validated_urls)} unique URL(s)\")\n\n if not unique_validated_urls:\n # Surface the actual reason every candidate URL was rejected (e.g. an\n # SSRF block) rather than a generic message, so a security failure is\n # not silently swallowed when continue_on_failure is enabled.\n if first_validation_error is not None:\n raise first_validation_error\n msg = \"No valid URLs provided.\"\n raise ValueError(msg)\n\n # Prepare headers\n headers_dict = {header[\"key\"]: header[\"value\"] for header in self.headers if header[\"value\"] is not None}\n\n # Crawl all URLs\n all_docs = []\n for url, validated_ips in unique_validated_urls:\n logger.debug(f\"Crawling {url} with max_depth={self.max_depth}\")\n\n try:\n visited = set()\n docs = await self._crawl_recursive(url, validated_ips, headers_dict, visited, depth=0)\n\n if not docs:\n logger.warning(f\"No documents found for {url}\")\n continue\n\n logger.debug(f\"Found {len(docs)} document(s) from {url}\")\n all_docs.extend(docs)\n\n except httpx.HTTPError as e:\n if self.continue_on_failure:\n logger.warning(f\"Error loading documents from {url}: {e}\")\n continue\n msg = f\"Error loading documents from {url}: {e}\"\n raise ValueError(msg) from e\n\n if not all_docs:\n msg = \"No documents were successfully loaded from any URL\"\n raise ValueError(msg)\n\n # Convert to output format\n return [\n {\n \"text\": safe_convert(doc[\"page_content\"], clean_data=True),\n \"url\": doc[\"metadata\"].get(\"source\", \"\"),\n \"title\": doc[\"metadata\"].get(\"title\", \"\"),\n \"description\": doc[\"metadata\"].get(\"description\", \"\"),\n \"content_type\": doc[\"metadata\"].get(\"content_type\", \"\"),\n \"language\": doc[\"metadata\"].get(\"language\", \"\"),\n }\n for doc in all_docs\n ]\n\n except Exception as e:\n error_msg = e.message if hasattr(e, \"message\") else str(e)\n msg = f\"Error loading documents: {error_msg}\"\n logger.exception(msg)\n raise ValueError(msg) from e\n\n async def fetch_content(self) -> DataFrame:\n \"\"\"Convert the documents to a DataFrame.\"\"\"\n url_contents = await self.fetch_url_contents()\n return DataFrame(data=url_contents)\n\n async def fetch_content_as_message(self) -> Message:\n \"\"\"Convert the documents to a Message.\"\"\"\n url_contents = await self.fetch_url_contents()\n return Message(text=\"\\n\\n\".join([x[\"text\"] for x in url_contents]), data={\"data\": url_contents})\n" + "value": "import importlib.util\nimport io\nimport os\nimport re\nimport zlib\nfrom urllib.parse import urljoin, urlparse\n\nimport httpx\nfrom bs4 import BeautifulSoup\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.field_typing.range_spec import RangeSpec\nfrom lfx.helpers.data import safe_convert\nfrom lfx.io import BoolInput, DropdownInput, IntInput, MessageTextInput, Output, SliderInput, TableInput\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\nfrom lfx.utils.request_utils import get_user_agent\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, is_ssrf_protection_enabled, validate_and_resolve_url\nfrom lfx.utils.ssrf_transport import create_ssrf_protected_client, pin_host_for_url\n\n# Constants\nDEFAULT_TIMEOUT = 30\nDEFAULT_MAX_DEPTH = 1\nDEFAULT_FORMAT = \"Text\"\n\n# HTTP status codes that carry a redirect Location header (RFC 9110).\nREDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308})\n\n# Maximum number of redirects to follow. Each hop is re-validated for SSRF safety;\n# matches httpx's default and the API Request component.\nMAX_REDIRECTS = 20\n\n# Default ports per scheme, used to compare redirect origins.\nDEFAULT_SCHEME_PORTS = {\"http\": 80, \"https\": 443}\n\n# Memory bounds for untrusted responses. Bodies are streamed and abandoned once the\n# per-response cap or the budget shared by every page of one fetch (all URLs and crawled\n# links) is exceeded, so a hostile server cannot exhaust memory with a huge or endless body.\n# Configurable via LANGFLOW_URL_COMPONENT_MAX_RESPONSE_BYTES / LANGFLOW_URL_COMPONENT_MAX_TOTAL_BYTES\n# (see RuntimeSettings); these are only the fallback used when the settings service isn't available.\nFALLBACK_MAX_RESPONSE_BYTES = 10 * 1024 * 1024\nFALLBACK_MAX_TOTAL_BYTES = 100 * 1024 * 1024\n\n# Only advertise codings we can decode with a strict output limit. Raw response bytes\n# are also capped before decoding, including compressed headers and trailing data.\nACCEPT_ENCODING = \"gzip, deflate\"\nALLOWED_CONTENT_ENCODINGS = frozenset({\"gzip\", \"deflate\"})\nZLIB_HEADER_BYTES = 2\n\n\nURL_REGEX = re.compile(\n r\"^(https?:\\/\\/)?\" r\"(www\\.)?\" r\"([a-zA-Z0-9.-]+)\" r\"(\\.[a-zA-Z]{2,})?\" r\"(:\\d+)?\" r\"(\\/[^\\s]*)?$\",\n re.IGNORECASE,\n)\n\nUSER_AGENT = None\n# Check if langflow is installed using importlib.util.find_spec(name))\nif importlib.util.find_spec(\"langflow\"):\n langflow_installed = True\n USER_AGENT = get_user_agent()\nelse:\n langflow_installed = False\n USER_AGENT = \"lfx\"\n\n\nclass URLComponent(Component):\n \"\"\"A component that loads and parses content from web pages recursively.\n\n This component allows fetching content from one or more URLs, with options to:\n - Control crawl depth\n - Prevent crawling outside the root domain\n - Use async loading for better performance\n - Extract either raw HTML or clean text\n - Configure request headers and timeouts\n \"\"\"\n\n display_name = \"URL\"\n description = \"Fetch content from one or more web pages, following links recursively.\"\n documentation: str = \"https://docs.langflow.org/url\"\n icon = \"layout-template\"\n name = \"URLComponent\"\n # Body bytes the current fetch may still read; reset by ``fetch_url_contents``.\n _bytes_remaining = FALLBACK_MAX_TOTAL_BYTES\n # Per-response and total byte caps for the current fetch; resolved from settings by\n # ``fetch_url_contents``. Kept as fallbacks here so the class works if that never runs.\n _max_response_bytes = FALLBACK_MAX_RESPONSE_BYTES\n _max_total_bytes = FALLBACK_MAX_TOTAL_BYTES\n\n inputs = [\n MessageTextInput(\n name=\"urls\",\n display_name=\"URLs\",\n info=\"Enter one or more URLs to crawl recursively, by clicking the '+' button.\",\n is_list=True,\n tool_mode=True,\n placeholder=\"Enter a URL...\",\n list_add_label=\"Add URL\",\n input_types=[\"Message\"],\n ),\n SliderInput(\n name=\"max_depth\",\n display_name=\"Depth\",\n info=(\n \"Controls how many 'clicks' away from the initial page the crawler will go:\\n\"\n \"- depth 1: only the initial page\\n\"\n \"- depth 2: initial page + all pages linked directly from it\\n\"\n \"- depth 3: initial page + direct links + links found on those direct link pages\\n\"\n \"Note: This is about link traversal, not URL path depth.\"\n ),\n value=DEFAULT_MAX_DEPTH,\n range_spec=RangeSpec(min=1, max=5, step=1),\n required=False,\n min_label=\" \",\n max_label=\" \",\n min_label_icon=\"None\",\n max_label_icon=\"None\",\n # slider_input=True\n ),\n BoolInput(\n name=\"prevent_outside\",\n display_name=\"Prevent Outside\",\n info=(\n \"If enabled, only crawls URLs within the same domain as the root URL. \"\n \"This helps prevent the crawler from going to external websites.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"use_async\",\n display_name=\"Use Async\",\n info=(\n \"If enabled, uses asynchronous loading which can be significantly faster \"\n \"but might use more system resources.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"follow_redirects\",\n display_name=\"Follow Redirects\",\n info=(\n \"If enabled, follows HTTP redirects such as httpβ†’https or www/non-www \"\n \"normalization, which most sites rely on to serve their content. When SSRF \"\n \"protection is enabled (the default), every redirect hop is re-validated against \"\n \"the same blocked-IP denylist and DNS-pinned before it is fetched, so following \"\n \"redirects cannot be used to reach internal resources. Disable to capture the \"\n \"first response as-is without following redirects.\"\n ),\n value=True,\n required=False,\n advanced=True,\n ),\n DropdownInput(\n name=\"format\",\n display_name=\"Output Format\",\n info=(\n \"Output Format. Use 'Text' to extract the text from the HTML, \"\n \"'Markdown' to parse the HTML into Markdown format, or 'HTML' \"\n \"for the raw HTML content.\"\n ),\n options=[\"Text\", \"HTML\", \"Markdown\"],\n value=DEFAULT_FORMAT,\n advanced=True,\n ),\n IntInput(\n name=\"timeout\",\n display_name=\"Timeout\",\n info=\"Timeout for the request in seconds.\",\n value=DEFAULT_TIMEOUT,\n required=False,\n advanced=True,\n ),\n TableInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=\"The headers to send with the request\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Header\",\n \"type\": \"str\",\n \"description\": \"Header name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"type\": \"str\",\n \"description\": \"Header value\",\n },\n ],\n value=[{\"key\": \"User-Agent\", \"value\": USER_AGENT}],\n advanced=True,\n input_types=[\"DataFrame\", \"Table\"],\n ),\n BoolInput(\n name=\"filter_text_html\",\n display_name=\"Filter Text/HTML\",\n info=\"If enabled, filters out text/css content type from the results.\",\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"continue_on_failure\",\n display_name=\"Continue on Failure\",\n info=\"If enabled, continues crawling even if some requests fail.\",\n value=True,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"check_response_status\",\n display_name=\"Check Response Status\",\n info=\"If enabled, checks the response status of the request.\",\n value=False,\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"autoset_encoding\",\n display_name=\"Autoset Encoding\",\n info=\"If enabled, automatically sets the encoding of the request.\",\n value=True,\n required=False,\n advanced=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Extracted Pages\", name=\"page_results\", method=\"fetch_content\"),\n Output(display_name=\"Raw Content\", name=\"raw_results\", method=\"fetch_content_as_message\", tool_mode=False),\n ]\n\n @staticmethod\n def _html_extractor(x: str) -> str:\n \"\"\"Extract raw HTML content.\"\"\"\n return x\n\n @staticmethod\n def _text_extractor(x: str) -> str:\n \"\"\"Extract clean text from HTML.\"\"\"\n return BeautifulSoup(x, \"lxml\").get_text()\n\n @staticmethod\n def _markdown_extractor(x: str) -> str:\n \"\"\"Convert HTML to Markdown format.\"\"\"\n # markitdown imports magika, which loads the nearest .env at import time. Keep that\n # optional converter out of component discovery and load it only when Markdown output\n # is actually requested.\n from markitdown import MarkItDown\n\n stream = io.BytesIO(x.encode(\"utf-8\"))\n result = MarkItDown(enable_plugins=False).convert_stream(stream)\n return result.markdown\n\n @staticmethod\n def validate_url(url: str) -> bool:\n \"\"\"Validates if the given string matches URL pattern.\n\n Args:\n url: The URL string to validate\n\n Returns:\n bool: True if the URL is valid, False otherwise\n \"\"\"\n return bool(URL_REGEX.match(url))\n\n def ensure_url(self, url: str) -> tuple[str, list[str]]:\n \"\"\"Ensures the given string is a valid URL and returns validated IPs for DNS pinning.\n\n Args:\n url: The URL string to validate and normalize\n\n Returns:\n tuple[str, list[str]]: The normalized URL and list of validated IPs for DNS pinning\n\n Raises:\n ValueError: If the URL is invalid or blocked by SSRF protection\n \"\"\"\n url = url.strip()\n if not url.startswith((\"http://\", \"https://\")):\n url = \"https://\" + url\n\n if not self.validate_url(url):\n msg = f\"Invalid URL: {url}\"\n raise ValueError(msg)\n\n # ============================================================================\n # SSRF Protection with DNS Pinning\n # ============================================================================\n # This prevents DNS rebinding attacks by:\n # 1. Resolving DNS and validating IPs during security check\n # 2. Returning the validated IP addresses for DNS pinning\n # 3. Using a custom HTTP transport that forces use of the pinned IPs\n # 4. Ignoring any new DNS resolutions (prevents rebinding)\n #\n # Without DNS pinning, an attacker could:\n # - First DNS lookup: returns public IP (passes validation)\n # - Second DNS lookup: returns internal IP (bypasses protection)\n # - Attack succeeds: accesses internal services\n #\n # With DNS pinning:\n # - First DNS lookup: returns public IP (passes validation)\n # - IPs are pinned and returned\n # - HTTP request: uses pinned IPs directly (no new DNS lookup)\n # - Attack fails: even if DNS changes, we use the validated IPs\n # ============================================================================\n try:\n _validated_url, validated_ips = validate_and_resolve_url(url)\n except SSRFProtectionError as e:\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n else:\n # Log DNS pinning information for security auditing\n if validated_ips:\n logger.debug(f\"SSRF Protection: Using DNS pinning with {len(validated_ips)} validated IP(s) for {url}\")\n\n return url, validated_ips\n\n def _build_http_client(self, url: str, validated_ips: list[str]) -> httpx.AsyncClient:\n \"\"\"Create an HTTP client with DNS pinning for SSRF protection.\n\n Args:\n url: The request URL whose hostname will be pinned\n validated_ips: IPs validated by validate_and_resolve_url for this URL\n\n Returns:\n httpx.AsyncClient: A client with DNS pinning when SSRF protection is enabled\n \"\"\"\n hostname = pin_host_for_url(url) if is_ssrf_protection_enabled() and validated_ips else None\n if hostname:\n client = create_ssrf_protected_client(hostname=hostname, validated_ips=validated_ips)\n else:\n client = httpx.AsyncClient()\n # A client default, so a user-supplied Accept-Encoding header still takes precedence.\n client.headers[\"Accept-Encoding\"] = ACCEPT_ENCODING\n return client\n\n @staticmethod\n def _headers_for_redirect(headers: dict | None, current_url: str, next_url: str) -> dict | None:\n \"\"\"Drop sensitive headers when a redirect crosses to a different origin.\n\n Mirrors httpx's auto-follow behavior so manually following redirects does not\n leak credentials (Authorization / Cookie) to an origin other than the one the\n caller intended them for. Headers are kept only when the redirect stays on the\n same origin (scheme, host, port) or is a direct https upgrade of the same host\n on default ports - the exact cases where httpx keeps the Authorization header.\n \"\"\"\n if not headers:\n return headers\n current, nxt = urlparse(current_url), urlparse(next_url)\n current_port = current.port or DEFAULT_SCHEME_PORTS.get(current.scheme)\n next_port = nxt.port or DEFAULT_SCHEME_PORTS.get(nxt.scheme)\n same_origin = (current.scheme, current.hostname, current_port) == (nxt.scheme, nxt.hostname, next_port)\n https_upgrade = (\n current.hostname == nxt.hostname\n and current.scheme == \"http\"\n and nxt.scheme == \"https\"\n and current_port == DEFAULT_SCHEME_PORTS[\"http\"]\n and next_port == DEFAULT_SCHEME_PORTS[\"https\"]\n )\n if same_origin or https_upgrade:\n return headers\n sensitive = {\"authorization\", \"proxy-authorization\", \"cookie\"}\n return {k: v for k, v in headers.items() if k.lower() not in sensitive}\n\n @staticmethod\n def _resolve_byte_limit(env_var: str, setting_name: str, fallback: int) -> int:\n \"\"\"Resolve one byte cap from settings, or the environment when no service is available.\n\n The settings service applies Langflow's normal configuration precedence. Reading the\n process environment directly is only a fallback for standalone ``lfx`` execution.\n \"\"\"\n try:\n settings_service = get_settings_service()\n except Exception: # noqa: BLE001 - service registry may not be ready\n settings_service = None\n if settings_service is not None:\n return getattr(settings_service.settings, setting_name, fallback)\n\n env_value = os.getenv(env_var)\n if env_value is not None:\n try:\n value = int(env_value)\n except ValueError:\n logger.warning(f\"Ignoring invalid {env_var}={env_value!r}; using {fallback}\")\n return fallback\n if value > 0:\n return value\n logger.warning(f\"Ignoring non-positive {env_var}={env_value!r}; using {fallback}\")\n return fallback\n\n return fallback\n\n @classmethod\n def _resolve_byte_limits(cls) -> tuple[int, int]:\n \"\"\"Read the configurable per-response and total byte caps for the current fetch.\"\"\"\n return (\n cls._resolve_byte_limit(\n \"LANGFLOW_URL_COMPONENT_MAX_RESPONSE_BYTES\",\n \"url_component_max_response_bytes\",\n FALLBACK_MAX_RESPONSE_BYTES,\n ),\n cls._resolve_byte_limit(\n \"LANGFLOW_URL_COMPONENT_MAX_TOTAL_BYTES\",\n \"url_component_max_total_bytes\",\n FALLBACK_MAX_TOTAL_BYTES,\n ),\n )\n\n async def _read_bounded_text(self, response: httpx.Response) -> str:\n \"\"\"Read a streamed response body without exceeding the per-response or total byte budget.\n\n Encoded and decoded bytes are each bounded. The larger count is charged against the\n total budget even when the body is rejected, so a server that keeps answering with\n oversized bodies cannot make a crawl download without bound. Decoder errors charge\n the reserved output bound because zlib may discard output before raising.\n\n Raises:\n httpx.HTTPError: If the body is (or declares to be) larger than the remaining budget,\n or uses a content coding whose decompression cannot be bounded.\n \"\"\"\n codings = [c.strip().lower() for c in response.headers.get(\"content-encoding\", \"\").split(\",\")]\n codings = [c for c in codings if c not in {\"\", \"identity\"}]\n if len(codings) > 1 or (codings and codings[0] not in ALLOWED_CONTENT_ENCODINGS):\n msg = f\"Unsupported content encoding from {response.url}: {', '.join(codings)}\"\n raise httpx.HTTPError(msg)\n\n limit = min(self._max_response_bytes, self._bytes_remaining)\n too_large = f\"Response from {response.url} exceeds the {limit} byte limit\"\n declared = response.headers.get(\"content-length\", \"\")\n if declared.isdigit() and int(declared) > limit:\n raise httpx.HTTPError(too_large)\n\n body = bytearray()\n raw_bytes = decoded_bytes = 0\n coding = codings[0] if codings else None\n decompressor = (\n zlib.decompressobj(zlib.MAX_WBITS | 16 if coding == \"gzip\" else zlib.MAX_WBITS) if coding else None\n )\n first_decode = True\n deflate_prefix = b\"\"\n try:\n # HTTPX's automatic decoder can allocate unbounded output or retain an\n # endless tail in zlib.unused_data without yielding any decoded bytes.\n async for raw_chunk in response.aiter_raw():\n raw_bytes += len(raw_chunk)\n if raw_bytes > limit:\n raise httpx.HTTPError(too_large)\n chunk = raw_chunk\n if decompressor is not None:\n if decompressor.eof:\n msg = f\"Trailing data after compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n if coding == \"deflate\" and first_decode:\n # HTTP deflate may omit the zlib wrapper. Wait for enough\n # header bytes before trying the same fallback as HTTPX.\n chunk = deflate_prefix + chunk\n if len(chunk) < ZLIB_HEADER_BYTES:\n deflate_prefix = chunk\n continue\n max_output = limit - decoded_bytes + 1\n try:\n try:\n decoded = decompressor.decompress(chunk, max_output)\n except zlib.error:\n if coding != \"deflate\" or not first_decode:\n raise\n decompressor = zlib.decompressobj(-zlib.MAX_WBITS)\n decoded = decompressor.decompress(chunk, max_output)\n except zlib.error as exc:\n # zlib can produce output before detecting a bad checksum,\n # then raise without returning it. Charge the reserved bound.\n decoded_bytes = limit + 1\n msg = f\"Invalid compressed response from {response.url}\"\n raise httpx.DecodingError(msg) from exc\n first_decode = False\n decoded_bytes += len(decoded)\n if decoded_bytes > limit:\n raise httpx.HTTPError(too_large)\n if decompressor.unused_data:\n msg = f\"Trailing data after compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n chunk = decoded\n else:\n decoded_bytes += len(chunk)\n # Check decoded output before retaining it in the response body.\n body.extend(chunk)\n if decompressor is not None and not decompressor.eof:\n msg = f\"Incomplete compressed response from {response.url}\"\n raise httpx.DecodingError(msg)\n finally:\n # Charge encoded overhead as well as expanded output, including rejected\n # responses. Neither representation may bypass the shared fetch budget.\n self._bytes_remaining -= max(raw_bytes, decoded_bytes)\n # Same character decoding as ``response.text``. A complete zlib stream has\n # already emitted all its output, so no unbounded decoder flush is needed.\n return body.decode(response.encoding or \"utf-8\", errors=\"replace\")\n\n async def _process_response(self, response: httpx.Response) -> tuple[str, dict]:\n \"\"\"Turn a final (non-redirect) streamed response into its HTML content and metadata.\n\n Args:\n response: The HTTP response to process, opened with ``stream=True``\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n if self.check_response_status:\n response.raise_for_status()\n\n content_type = response.headers.get(\"content-type\", \"\").lower()\n\n # Filter out CSS files if requested\n if self.filter_text_html and \"text/css\" in content_type:\n logger.debug(f\"Skipping CSS file: {response.url}\")\n return \"\", {}\n\n # Get the HTML content\n html_content = await self._read_bounded_text(response)\n\n # Extract metadata\n metadata = {\n \"source\": str(response.url),\n \"title\": \"\",\n \"description\": \"\",\n \"content_type\": content_type,\n \"language\": \"\",\n }\n\n # Try to extract title and description from HTML\n try:\n soup = BeautifulSoup(html_content, \"lxml\")\n except Exception: # noqa: BLE001\n # Broad exception is acceptable here - metadata extraction is optional\n # and we don't want to fail the entire request if it fails\n logger.debug(f\"Failed to extract metadata from {response.url}\")\n else:\n if soup.title:\n metadata[\"title\"] = soup.title.string or \"\"\n\n # Try to get description from meta tags\n meta_desc = soup.find(\"meta\", attrs={\"name\": \"description\"})\n if meta_desc and meta_desc.get(\"content\"):\n metadata[\"description\"] = meta_desc[\"content\"]\n\n # Try to get language\n html_tag = soup.find(\"html\")\n if html_tag and html_tag.get(\"lang\"):\n metadata[\"language\"] = html_tag[\"lang\"]\n\n return html_content, metadata\n\n async def _fetch_with_revalidated_redirects(\n self, url: str, validated_ips: list[str], headers: dict\n ) -> tuple[str, dict]:\n \"\"\"Fetch ``url``, following redirects manually and re-validating every hop.\n\n This closes an SSRF bypass: with redirects enabled, httpx would otherwise\n auto-follow a redirect from a validated public URL to an internal address that\n the pinned transport never checked (the redirect target is a different host, so\n it is not in the pin map). Each redirect ``Location`` is resolved (relative\n locations included) and re-validated with ``ensure_url`` - the same private/\n loopback/link-local denylist and DNS pinning applied to the initial request -\n before any connection to it is made. A blocked hop raises ``ValueError`` (unless\n ``continue_on_failure``); the chain is capped at ``MAX_REDIRECTS`` hops.\n\n Args:\n url: The URL to fetch\n validated_ips: Validated IPs for DNS pinning of the initial URL\n headers: HTTP headers to send\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n current_url = url\n current_ips = validated_ips\n\n for _ in range(MAX_REDIRECTS + 1):\n # Stream so redirect bodies are never read and the final body is read bounded.\n async with (\n self._build_http_client(current_url, current_ips) as client,\n client.stream(\n \"GET\", current_url, headers=headers, timeout=self.timeout, follow_redirects=False\n ) as response,\n ):\n location = response.headers.get(\"location\")\n if response.status_code not in REDIRECT_STATUS_CODES or not location:\n # Not a redirect (or no Location header) - this is the final response.\n return await self._process_response(response)\n\n # Resolve relative redirects against the current URL.\n next_url = urljoin(current_url, location)\n\n # Re-validate the redirect target with the same SSRF denylist + DNS pinning.\n try:\n validated_next_url, current_ips = self.ensure_url(next_url)\n except (ValueError, SSRFProtectionError) as e:\n if self.continue_on_failure:\n logger.warning(f\"Skipping blocked or invalid redirect to {next_url}: {e}\")\n return \"\", {}\n msg = f\"SSRF Protection: blocked redirect to {next_url}: {e}\"\n raise ValueError(msg) from e\n\n headers = self._headers_for_redirect(headers, current_url, next_url)\n current_url = validated_next_url\n\n # Exhausted the redirect budget.\n if self.continue_on_failure:\n logger.warning(f\"Exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}\")\n return \"\", {}\n msg = f\"SSRF Protection: exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}\"\n raise ValueError(msg)\n\n async def _fetch_url_with_pinning(self, url: str, validated_ips: list[str], headers: dict) -> tuple[str, dict]:\n \"\"\"Fetch a single URL with DNS pinning protection, following redirects safely.\n\n Args:\n url: The URL to fetch\n validated_ips: Validated IPs for DNS pinning\n headers: HTTP headers to send\n\n Returns:\n tuple[str, dict]: The HTML content and metadata\n \"\"\"\n try:\n # When SSRF protection is enabled we must follow redirects manually so each hop\n # is re-validated and DNS-pinned; letting httpx auto-follow would connect to the\n # redirect target without pinning (different host, not in the pin map) and re-open\n # the SSRF hole that DNS pinning closes. With protection disabled there is no pin\n # to bypass, so redirects follow the requests httpx itself builds for each hop.\n if self.follow_redirects and is_ssrf_protection_enabled():\n return await self._fetch_with_revalidated_redirects(url, validated_ips, headers)\n\n async with self._build_http_client(url, validated_ips) as client:\n request = client.build_request(\"GET\", url, headers=headers, timeout=self.timeout)\n # httpx auto-follow reads every redirect body into memory without a limit, so hop\n # manually via ``next_request`` (same cookies, headers and redirect cap as httpx).\n for _ in range(MAX_REDIRECTS + 1):\n response = await client.send(request, stream=True, follow_redirects=False)\n try:\n if not (self.follow_redirects and response.next_request):\n return await self._process_response(response)\n request = response.next_request\n finally:\n await response.aclose()\n msg = \"Exceeded maximum allowed redirects.\"\n raise httpx.TooManyRedirects(msg, request=request)\n\n except httpx.HTTPError as e:\n if self.continue_on_failure:\n logger.warning(f\"Failed to fetch {url}: {e}\")\n return \"\", {}\n raise\n\n async def _crawl_recursive(\n self, start_url: str, validated_ips: list[str], headers: dict, visited: set, depth: int = 0\n ) -> list[dict]:\n \"\"\"Recursively crawl URLs with DNS pinning protection.\n\n Args:\n start_url: The URL to start crawling from\n validated_ips: Validated IPs for DNS pinning\n headers: HTTP headers to send\n visited: Set of already visited URLs\n depth: Current crawl depth\n\n Returns:\n list[dict]: List of documents with content and metadata\n \"\"\"\n if depth >= self.max_depth or start_url in visited:\n return []\n if self._bytes_remaining <= 0:\n logger.warning(f\"Skipping {start_url}: reached the {self._max_total_bytes} byte limit for this fetch\")\n return []\n\n visited.add(start_url)\n documents = []\n\n # Fetch the current URL\n html_content, metadata = await self._fetch_url_with_pinning(start_url, validated_ips, headers)\n\n # Redirects may have landed on a different canonical URL (http->https, www\n # normalization). Resolve relative links and the prevent_outside check against\n # the URL the content actually came from, and mark it visited so links back to\n # the canonical form are not re-crawled.\n base_url = metadata.get(\"source\") or start_url\n visited.add(base_url)\n\n if not html_content:\n return documents\n\n # Extract content based on format\n extractors = {\n \"HTML\": self._html_extractor,\n \"Markdown\": self._markdown_extractor,\n \"Text\": self._text_extractor,\n }\n extractor = extractors.get(self.format, self._text_extractor)\n extracted_content = extractor(html_content)\n\n # Add the document\n documents.append(\n {\n \"page_content\": extracted_content,\n \"metadata\": metadata,\n }\n )\n\n # If we haven't reached max depth, extract and follow links\n if depth < self.max_depth - 1:\n try:\n soup = BeautifulSoup(html_content, \"lxml\")\n links = soup.find_all(\"a\", href=True)\n\n for link in links:\n if self._bytes_remaining <= 0:\n break\n href = link[\"href\"]\n # Resolve relative URLs\n absolute_url = urljoin(base_url, href)\n\n # Skip if already visited\n if absolute_url in visited:\n continue\n\n # Check if we should prevent going outside the domain\n if self.prevent_outside:\n start_domain = urlparse(base_url).netloc\n link_domain = urlparse(absolute_url).netloc\n if start_domain != link_domain:\n continue\n\n # Validate and crawl the linked URL\n try:\n _, link_validated_ips = self.ensure_url(absolute_url)\n sub_docs = await self._crawl_recursive(\n absolute_url, link_validated_ips, headers, visited, depth + 1\n )\n documents.extend(sub_docs)\n except (ValueError, SSRFProtectionError) as e:\n if self.continue_on_failure:\n logger.warning(f\"Skipping {absolute_url}: {e}\")\n continue\n raise\n\n except Exception: # noqa: BLE001\n # Broad exception is acceptable here - link extraction is optional\n # and we don't want to fail the entire crawl if one page has issues\n logger.debug(f\"Failed to extract links from {start_url}\")\n\n return documents\n\n async def fetch_url_contents(self) -> list[dict]:\n \"\"\"Load documents from the configured URLs with SSRF protection and DNS pinning.\n\n This method implements comprehensive SSRF (Server-Side Request Forgery) protection\n using DNS pinning to prevent DNS rebinding attacks. Each URL is validated and its\n DNS resolution is pinned before any HTTP requests are made.\n\n Returns:\n list[dict]: List of documents with content and metadata\n\n Raises:\n ValueError: If no valid URLs are provided or if there's an error loading documents\n \"\"\"\n self._max_response_bytes, self._max_total_bytes = self._resolve_byte_limits()\n self._bytes_remaining = self._max_total_bytes\n try:\n # Validate all URLs and get their validated IPs for DNS pinning\n validated_urls = []\n first_validation_error: Exception | None = None\n for url in self.urls:\n if not url.strip():\n continue\n try:\n normalized_url, validated_ips = self.ensure_url(url)\n validated_urls.append((normalized_url, validated_ips))\n except (ValueError, SSRFProtectionError) as e:\n # Remember the first rejection so its real cause (for example an\n # SSRF block) can be surfaced if no URL survives validation,\n # instead of being hidden behind \"No valid URLs provided\".\n if first_validation_error is None:\n first_validation_error = e\n if self.continue_on_failure:\n logger.warning(f\"Skipping invalid URL {url}: {e}\")\n continue\n raise\n\n # Remove duplicates while preserving order\n seen = set()\n unique_validated_urls = []\n for url, ips in validated_urls:\n if url not in seen:\n seen.add(url)\n unique_validated_urls.append((url, ips))\n\n logger.debug(f\"Validated {len(unique_validated_urls)} unique URL(s)\")\n\n if not unique_validated_urls:\n # Surface the actual reason every candidate URL was rejected (e.g. an\n # SSRF block) rather than a generic message, so a security failure is\n # not silently swallowed when continue_on_failure is enabled.\n if first_validation_error is not None:\n raise first_validation_error\n msg = \"No valid URLs provided.\"\n raise ValueError(msg)\n\n # Prepare headers\n headers_dict = {header[\"key\"]: header[\"value\"] for header in self.headers if header[\"value\"] is not None}\n\n # Crawl all URLs\n all_docs = []\n for url, validated_ips in unique_validated_urls:\n logger.debug(f\"Crawling {url} with max_depth={self.max_depth}\")\n\n try:\n visited = set()\n docs = await self._crawl_recursive(url, validated_ips, headers_dict, visited, depth=0)\n\n if not docs:\n logger.warning(f\"No documents found for {url}\")\n continue\n\n logger.debug(f\"Found {len(docs)} document(s) from {url}\")\n all_docs.extend(docs)\n\n except httpx.HTTPError as e:\n if self.continue_on_failure:\n logger.warning(f\"Error loading documents from {url}: {e}\")\n continue\n msg = f\"Error loading documents from {url}: {e}\"\n raise ValueError(msg) from e\n\n if not all_docs:\n msg = \"No documents were successfully loaded from any URL\"\n raise ValueError(msg)\n\n # Convert to output format\n return [\n {\n \"text\": safe_convert(doc[\"page_content\"], clean_data=True),\n \"url\": doc[\"metadata\"].get(\"source\", \"\"),\n \"title\": doc[\"metadata\"].get(\"title\", \"\"),\n \"description\": doc[\"metadata\"].get(\"description\", \"\"),\n \"content_type\": doc[\"metadata\"].get(\"content_type\", \"\"),\n \"language\": doc[\"metadata\"].get(\"language\", \"\"),\n }\n for doc in all_docs\n ]\n\n except Exception as e:\n error_msg = e.message if hasattr(e, \"message\") else str(e)\n msg = f\"Error loading documents: {error_msg}\"\n logger.exception(msg)\n raise ValueError(msg) from e\n\n async def fetch_content(self) -> DataFrame:\n \"\"\"Convert the documents to a DataFrame.\"\"\"\n url_contents = await self.fetch_url_contents()\n return DataFrame(data=url_contents)\n\n async def fetch_content_as_message(self) -> Message:\n \"\"\"Convert the documents to a Message.\"\"\"\n url_contents = await self.fetch_url_contents()\n return Message(text=\"\\n\\n\".join([x[\"text\"] for x in url_contents]), data={\"data\": url_contents})\n" }, "continue_on_failure": { "_input_type": "BoolInput", @@ -5766,7 +5766,7 @@ "icon": "file-text", "legacy": false, "metadata": { - "code_hash": "6d53589f01c8", + "code_hash": "1bc59574381f", "component_class_name": "FileComponent", "dependencies": { "dependencies": [ @@ -5935,7 +5935,7 @@ "show": true, "title_case": false, "type": "code", - "value": "\"\"\"Enhanced file component with Docling support and process isolation.\n\nNotes:\n-----\n- ALL Docling parsing/export runs in a separate OS process to prevent memory\n growth and native library state from impacting the main Langflow process.\n- Standard text/structured parsing continues to use existing BaseFileComponent\n utilities, bounded by `concurrency_multithreading` via `aparallel_load_data`.\n- The loaders are coroutines (see BaseFileComponent): storage reads are awaited on\n the caller's event loop, parsing and the Docling subprocess run in worker threads.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport contextlib\nimport json\nimport subprocess\nimport sys\nimport textwrap\nimport threading\nimport time\nfrom contextvars import ContextVar\nfrom copy import deepcopy\nfrom pathlib import Path\nfrom tempfile import NamedTemporaryFile\nfrom typing import Any\n\nfrom lfx.base.data.base_file import BaseFileComponent\nfrom lfx.base.data.storage_utils import (\n parse_storage_path,\n read_file_bytes,\n require_storage_service,\n validate_image_content_type,\n)\nfrom lfx.base.data.utils import TEXT_FILE_TYPES, aparallel_load_data, aparse_text_file_to_data\nfrom lfx.inputs import SortableListInput\nfrom lfx.inputs.inputs import DropdownInput, MessageTextInput, StrInput\nfrom lfx.io import BoolInput, FileInput, IntInput, Output, SecretStrInput\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame # noqa: TC001\nfrom lfx.schema.message import Message\nfrom lfx.services.deps import get_settings_service, get_storage_service\nfrom lfx.utils.async_helpers import delegates_to, run_until_complete\nfrom lfx.utils.validate_cloud import is_astra_cloud_environment\n\n_FILE_TOOL_CANCEL_EVENT: ContextVar[threading.Event | None] = ContextVar(\"file_tool_cancel_event\", default=None)\n_FILE_TOOL_CANCEL_WAIT_SECONDS = 6\n_FILE_TOOL_MAX_CONCURRENT_LOADS = 4\n_FILE_TOOL_LIMITER_ATTRIBUTE = \"_lfx_file_tool_load_limiter\"\n_FILE_TOOL_PROCESS_REAP_SECONDS = 2\n\n\nclass _FileToolCancelledError(RuntimeError):\n \"\"\"Stop synchronous file loading after its async tool call is cancelled.\"\"\"\n\n\ndef _raise_if_file_tool_cancelled(cancel_event: threading.Event | None = None) -> None:\n event = cancel_event or _FILE_TOOL_CANCEL_EVENT.get()\n if event is not None and event.is_set():\n raise _FileToolCancelledError\n\n\ndef _kill_and_reap_process(proc: subprocess.Popen) -> None:\n try:\n proc.kill()\n except ProcessLookupError:\n pass\n except OSError:\n logger.exception(\"Failed to kill cancelled Docling subprocess\")\n except Exception: # noqa: BLE001 - cleanup must not mask the caller's cancellation\n logger.exception(\"Unexpected error while killing cancelled Docling subprocess\")\n\n try:\n proc.communicate(timeout=_FILE_TOOL_PROCESS_REAP_SECONDS)\n except subprocess.TimeoutExpired:\n logger.warning(\"Timed out draining cancelled Docling subprocess; waiting for process exit\")\n except OSError:\n logger.exception(\"Failed to drain cancelled Docling subprocess\")\n except Exception: # noqa: BLE001 - cleanup must not mask the caller's cancellation\n logger.exception(\"Unexpected error while draining cancelled Docling subprocess\")\n else:\n return\n\n try:\n proc.wait(timeout=_FILE_TOOL_PROCESS_REAP_SECONDS)\n except subprocess.TimeoutExpired:\n logger.exception(\"Cancelled Docling subprocess could not be reaped within the cleanup timeout\")\n except OSError:\n logger.exception(\"Failed to reap cancelled Docling subprocess\")\n except Exception: # noqa: BLE001 - cleanup must not mask the caller's cancellation\n logger.exception(\"Unexpected error while reaping cancelled Docling subprocess\")\n\n\ndef _get_file_tool_limiter() -> asyncio.Semaphore:\n \"\"\"Return the bounded loader admission gate for the current event loop.\"\"\"\n loop = asyncio.get_running_loop()\n limiter = getattr(loop, _FILE_TOOL_LIMITER_ATTRIBUTE, None)\n if limiter is None:\n limiter = asyncio.Semaphore(_FILE_TOOL_MAX_CONCURRENT_LOADS)\n setattr(loop, _FILE_TOOL_LIMITER_ATTRIBUTE, limiter)\n return limiter\n\n\ndef _log_abandoned_file_loader_result(task: asyncio.Task) -> None:\n \"\"\"Observe unexpected failures from a synchronous loader that outlived its caller.\"\"\"\n try:\n error = task.exception()\n except asyncio.CancelledError:\n return\n if error is not None and not isinstance(error, _FileToolCancelledError):\n logger.error(\"Abandoned file loader failed after its caller was cancelled\", exc_info=error)\n\n\ndef _write_temp_file(content: bytes, suffix: str) -> str:\n with NamedTemporaryFile(mode=\"wb\", suffix=suffix, delete=False) as tmp_file:\n tmp_file.write(content)\n return tmp_file.name\n\n\ndef _unlink_orphaned_temp_file(write: asyncio.Future[str]) -> None:\n \"\"\"Remove a temp file whose writer finished after the awaiting coroutine was cancelled.\"\"\"\n if write.cancelled() or write.exception() is not None:\n return\n with contextlib.suppress(OSError):\n Path(write.result()).unlink()\n\n\nasync def _wait_for_cancelled_file_tool_loader(loader_task: asyncio.Task) -> None:\n \"\"\"Give a cancelled tool call's loader time to clean up, without cancelling it.\n\n This waits with ``asyncio.wait`` rather than ``asyncio.shield``. On Python 3.14, a\n shield whose outer future is cancelled reports the inner task's eventual exception to\n the loop's exception handler, which logged the expected ``_FileToolCancelledError`` as\n an ERROR on every cancelled tool call.\n \"\"\"\n try:\n await asyncio.wait({loader_task}, timeout=_FILE_TOOL_CANCEL_WAIT_SECONDS)\n finally:\n if not loader_task.done():\n loader_task.add_done_callback(_log_abandoned_file_loader_result)\n elif not loader_task.cancelled():\n error = loader_task.exception()\n if error is not None and not isinstance(error, _FileToolCancelledError):\n logger.error(\"File loader failed while cleaning up a cancelled tool call\", exc_info=error)\n\n\ndef _get_storage_location_options():\n \"\"\"Get storage location options, filtering out Local if in Astra cloud environment.\"\"\"\n all_options = [{\"name\": \"AWS\", \"icon\": \"Amazon\"}, {\"name\": \"Google Drive\", \"icon\": \"google\"}]\n if is_astra_cloud_environment():\n return all_options\n return [{\"name\": \"Local\", \"icon\": \"hard-drive\"}, *all_options]\n\n\nclass FileComponent(BaseFileComponent):\n \"\"\"File component with optional Docling processing (isolated in a subprocess).\"\"\"\n\n display_name = \"Read File\"\n # description is now a dynamic property - see get_tool_description()\n _base_description = \"Loads and returns the content from uploaded files.\"\n documentation: str = \"https://docs.langflow.org/read-file\"\n icon = \"file-text\"\n name = \"File\"\n add_tool_output = True # Enable tool mode toggle without requiring tool_mode inputs\n\n # Extensions that can be processed without Docling (using standard text parsing)\n TEXT_EXTENSIONS = TEXT_FILE_TYPES\n\n # Extensions that require Docling for processing (images, advanced office formats, etc.)\n DOCLING_ONLY_EXTENSIONS = [\n \"adoc\",\n \"asciidoc\",\n \"asc\",\n \"bmp\",\n \"dotx\",\n \"dotm\",\n \"docm\",\n \"jpg\",\n \"jpeg\",\n \"png\",\n \"potx\",\n \"ppsx\",\n \"pptm\",\n \"potm\",\n \"ppsm\",\n \"pptx\",\n \"tiff\",\n \"xls\",\n \"xlsx\",\n \"xhtml\",\n \"webp\",\n ]\n\n # Docling-supported/compatible extensions; TEXT_FILE_TYPES are supported by the base loader.\n VALID_EXTENSIONS = [\n *TEXT_EXTENSIONS,\n *DOCLING_ONLY_EXTENSIONS,\n ]\n\n # Fixed export settings used when markdown export is requested.\n EXPORT_FORMAT = \"Markdown\"\n IMAGE_MODE = \"placeholder\"\n\n _base_inputs = deepcopy(BaseFileComponent.get_base_inputs())\n\n for input_item in _base_inputs:\n if isinstance(input_item, FileInput) and input_item.name == \"path\":\n input_item.real_time_refresh = True\n input_item.tool_mode = False # Disable tool mode for file upload input\n input_item.required = False # Make it optional so it doesn't error in tool mode\n break\n\n inputs = [\n SortableListInput(\n name=\"storage_location\",\n display_name=\"Storage Location\",\n placeholder=\"Select Location\",\n info=\"Choose where to read the file from.\",\n options=_get_storage_location_options(),\n real_time_refresh=True,\n limit=1,\n value=[{\"name\": \"Local\", \"icon\": \"hard-drive\"}],\n advanced=True,\n ),\n *_base_inputs,\n StrInput(\n name=\"text_encoding\",\n display_name=\"Text Encoding\",\n value=\"\",\n advanced=True,\n info=(\n \"Leave blank to detect text encoding automatically. If accents or punctuation look incorrect, \"\n \"enter the file's encoding, such as cp1252 or utf-8. Applies to standard text parsing and CSV, \"\n \"not PDF, DOCX, or the Advanced Parser. The selected encoding is shown in logs and output metadata.\"\n ),\n ),\n StrInput(\n name=\"file_path_str\",\n display_name=\"File Path\",\n info=(\n \"Path to the file to read. Used when component is called as a tool. \"\n \"If not provided, will use the uploaded file from 'path' input.\"\n ),\n show=False,\n advanced=True,\n tool_mode=True, # Required for Toolset toggle, but _get_tools() ignores this parameter\n required=False,\n ),\n # AWS S3 specific inputs\n SecretStrInput(\n name=\"aws_access_key_id\",\n display_name=\"AWS Access Key ID\",\n info=\"AWS Access key ID.\",\n show=False,\n advanced=False,\n required=True,\n ),\n SecretStrInput(\n name=\"aws_secret_access_key\",\n display_name=\"AWS Secret Key\",\n info=\"AWS Secret Key.\",\n show=False,\n advanced=False,\n required=True,\n ),\n StrInput(\n name=\"bucket_name\",\n display_name=\"S3 Bucket Name\",\n info=\"Enter the name of the S3 bucket.\",\n show=False,\n advanced=False,\n required=True,\n ),\n StrInput(\n name=\"aws_region\",\n display_name=\"AWS Region\",\n info=\"AWS region (e.g., us-east-1, eu-west-1).\",\n show=False,\n advanced=False,\n ),\n StrInput(\n name=\"s3_file_key\",\n display_name=\"S3 File Key\",\n info=\"The key (path) of the file in S3 bucket.\",\n show=False,\n advanced=False,\n required=True,\n ),\n # Google Drive specific inputs\n SecretStrInput(\n name=\"service_account_key\",\n display_name=\"GCP Credentials Secret Key\",\n info=\"Your Google Cloud Platform service account JSON key as a secret string (complete JSON content).\",\n show=False,\n advanced=False,\n required=True,\n ),\n StrInput(\n name=\"file_id\",\n display_name=\"Google Drive File ID\",\n info=(\"The Google Drive file ID to read. The file must be shared with the service account email.\"),\n show=False,\n advanced=False,\n required=True,\n ),\n BoolInput(\n name=\"advanced_mode\",\n display_name=\"Advanced Parser\",\n value=False,\n real_time_refresh=True,\n info=(\n \"Enable advanced document processing and export with Docling for PDFs, images, and office documents. \"\n \"Note that advanced document processing can consume significant resources.\"\n ),\n # Disabled in cloud\n show=not is_astra_cloud_environment(),\n ),\n DropdownInput(\n name=\"pipeline\",\n display_name=\"Pipeline\",\n info=\"Docling pipeline to use\",\n options=[\"standard\", \"vlm\"],\n value=\"standard\",\n advanced=True,\n real_time_refresh=True,\n ),\n DropdownInput(\n name=\"ocr_engine\",\n display_name=\"OCR Engine\",\n info=\"OCR engine to use. Only available when pipeline is set to 'standard'.\",\n options=[\"None\", \"easyocr\"],\n value=\"easyocr\",\n show=False,\n advanced=True,\n ),\n StrInput(\n name=\"md_image_placeholder\",\n display_name=\"Image placeholder\",\n info=\"Specify the image placeholder for markdown exports.\",\n value=\"\",\n advanced=True,\n show=False,\n ),\n StrInput(\n name=\"md_page_break_placeholder\",\n display_name=\"Page break placeholder\",\n info=\"Add this placeholder between pages in the markdown output.\",\n value=\"\",\n advanced=True,\n show=False,\n ),\n MessageTextInput(\n name=\"doc_key\",\n display_name=\"Doc Key\",\n info=\"The key to use for the DoclingDocument column.\",\n value=\"doc\",\n advanced=True,\n show=False,\n ),\n # Deprecated input retained for backward-compatibility.\n BoolInput(\n name=\"use_multithreading\",\n display_name=\"[Deprecated] Use Multithreading\",\n advanced=True,\n value=True,\n info=\"Set 'Processing Concurrency' greater than 1 to enable multithreading.\",\n ),\n IntInput(\n name=\"concurrency_multithreading\",\n display_name=\"Processing Concurrency\",\n advanced=True,\n info=\"When multiple files are being processed, the number of files to process concurrently.\",\n value=1,\n ),\n BoolInput(\n name=\"markdown\",\n display_name=\"Markdown Export\",\n info=\"Export processed documents to Markdown format. Only available when advanced mode is enabled.\",\n value=False,\n show=False,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Raw Content\", name=\"message\", method=\"load_files_message\", tool_mode=True),\n ]\n\n # ------------------------------ Tool description with file names --------------\n\n def get_tool_description(self) -> str:\n \"\"\"Return a dynamic description that includes the names of uploaded files.\n\n This helps the Agent understand which files are available to read.\n \"\"\"\n base_description = type(self)._base_description # noqa: SLF001\n\n # Get the list of uploaded file paths\n file_paths = getattr(self, \"path\", None)\n if not file_paths:\n return base_description\n\n # Ensure it's a list\n if not isinstance(file_paths, list):\n file_paths = [file_paths]\n\n # Extract just the file names from the paths\n file_names = []\n for fp in file_paths:\n if fp:\n name = Path(fp).name\n file_names.append(name)\n\n if file_names:\n files_str = \", \".join(file_names)\n return f\"{base_description} Available files: {files_str}. Call this tool to read these files.\"\n\n return base_description\n\n @property\n def description(self) -> str:\n \"\"\"Dynamic description property that includes uploaded file names.\"\"\"\n return self.get_tool_description()\n\n async def _get_tools(self) -> list:\n \"\"\"Override to create a tool without parameters.\n\n The Read File component should use the files already uploaded via UI,\n not accept file paths from the Agent (which wouldn't know the internal paths).\n \"\"\"\n from langchain_core.tools import StructuredTool\n from pydantic import BaseModel\n\n # Empty schema - no parameters needed\n class EmptySchema(BaseModel):\n \"\"\"No parameters required - uses pre-uploaded files.\"\"\"\n\n async def read_files_tool() -> str:\n \"\"\"Read the content of uploaded files.\"\"\"\n cancel_event = threading.Event()\n cancel_token = _FILE_TOOL_CANCEL_EVENT.set(cancel_event)\n try:\n if getattr(self, \"advanced_mode\", False):\n # In advanced mode, use the markdown output path so that the\n # tool shares the same Docling processing as the advanced\n # outputs rather than triggering a second subprocess via\n # load_files_message.\n self.markdown = True\n loader_name = \"load_files_markdown\"\n else:\n loader_name = \"load_files_message\"\n # Await the async loader chain on this loop: storage reads are awaited here,\n # while parsing and the Docling subprocess run in worker threads, so\n # streaming/heartbeats on the same loop keep flowing while the agent waits.\n # A subclass that overrides the sync loader still has it run in a thread.\n # Keep admission bounded until the real work exits so cancelled parsers\n # cannot build an unbounded default-executor backlog.\n load_limiter = _get_file_tool_limiter()\n await load_limiter.acquire()\n try:\n loader_task = asyncio.create_task(self._adispatch(loader_name))\n except BaseException:\n load_limiter.release()\n raise\n loader_task.add_done_callback(lambda _task: load_limiter.release())\n try:\n # asyncio.wait neither cancels the loader nor leaves an orphaned\n # shield behind to report its outcome if this call is cancelled.\n await asyncio.wait({loader_task})\n # A cancelled loader task raises CancelledError here while its\n # worker thread keeps running, so it takes the same path below.\n result = loader_task.result()\n except asyncio.CancelledError:\n cancel_event.set()\n # Give cooperative cleanup time to kill/reap Docling and remove\n # temporary files before releasing this tool invocation.\n await _wait_for_cancelled_file_tool_loader(loader_task)\n raise\n if hasattr(result, \"get_text\"):\n return result.get_text()\n if hasattr(result, \"text\"):\n return result.text\n return str(result)\n except (FileNotFoundError, ValueError, OSError, RuntimeError) as e:\n return f\"Error reading files: {e}\"\n finally:\n _FILE_TOOL_CANCEL_EVENT.reset(cancel_token)\n\n description = self.get_tool_description()\n\n tool = StructuredTool(\n name=\"load_files_message\",\n description=description,\n coroutine=read_files_tool,\n args_schema=EmptySchema,\n handle_tool_error=True,\n tags=[\"load_files_message\"],\n metadata={\n \"display_name\": \"Read File\",\n \"display_description\": description,\n },\n )\n\n return [tool]\n\n # ------------------------------ UI helpers --------------------------------------\n\n def _path_value(self, template: dict) -> list[str]:\n \"\"\"Return the list of currently selected file paths from the template.\"\"\"\n return template.get(\"path\", {}).get(\"file_path\", [])\n\n def _disable_docling_fields_in_cloud(self, build_config: dict[str, Any]) -> None:\n \"\"\"Disable all Docling-related fields in cloud environments.\"\"\"\n if \"advanced_mode\" in build_config:\n build_config[\"advanced_mode\"][\"show\"] = False\n build_config[\"advanced_mode\"][\"value\"] = False\n # Hide all Docling-related fields\n docling_fields = (\"pipeline\", \"ocr_engine\", \"doc_key\", \"md_image_placeholder\", \"md_page_break_placeholder\")\n for field in docling_fields:\n if field in build_config:\n build_config[field][\"show\"] = False\n # Also disable OCR engine specifically\n if \"ocr_engine\" in build_config:\n build_config[\"ocr_engine\"][\"value\"] = \"None\"\n\n def update_build_config(\n self,\n build_config: dict[str, Any],\n field_value: Any,\n field_name: str | None = None,\n ) -> dict[str, Any]:\n \"\"\"Show/hide Advanced Parser and related fields based on selection context.\"\"\"\n # Update storage location options dynamically based on cloud environment\n if \"storage_location\" in build_config:\n updated_options = _get_storage_location_options()\n build_config[\"storage_location\"][\"options\"] = updated_options\n\n # Handle storage location selection\n if field_name == \"storage_location\":\n # Extract selected storage location\n selected = [location[\"name\"] for location in field_value] if isinstance(field_value, list) else []\n\n # Hide all storage-specific fields first\n storage_fields = [\n \"aws_access_key_id\",\n \"aws_secret_access_key\",\n \"bucket_name\",\n \"aws_region\",\n \"s3_file_key\",\n \"service_account_key\",\n \"file_id\",\n ]\n\n for f_name in storage_fields:\n if f_name in build_config:\n build_config[f_name][\"show\"] = False\n\n # Show fields based on selected storage location\n if len(selected) == 1:\n location = selected[0]\n\n if location == \"Local\":\n # Show file upload input for local storage\n if \"path\" in build_config:\n build_config[\"path\"][\"show\"] = True\n\n elif location == \"AWS\":\n # Hide file upload input, show AWS fields\n if \"path\" in build_config:\n build_config[\"path\"][\"show\"] = False\n\n aws_fields = [\n \"aws_access_key_id\",\n \"aws_secret_access_key\",\n \"bucket_name\",\n \"aws_region\",\n \"s3_file_key\",\n ]\n for f_name in aws_fields:\n if f_name in build_config:\n build_config[f_name][\"show\"] = True\n build_config[f_name][\"advanced\"] = False\n\n elif location == \"Google Drive\":\n # Hide file upload input, show Google Drive fields\n if \"path\" in build_config:\n build_config[\"path\"][\"show\"] = False\n\n gdrive_fields = [\"service_account_key\", \"file_id\"]\n for f_name in gdrive_fields:\n if f_name in build_config:\n build_config[f_name][\"show\"] = True\n build_config[f_name][\"advanced\"] = False\n # No storage location selected - show file upload by default\n elif \"path\" in build_config:\n build_config[\"path\"][\"show\"] = True\n\n return build_config\n\n if field_name == \"path\":\n paths = self._path_value(build_config)\n\n # Disable in cloud environments\n if is_astra_cloud_environment():\n self._disable_docling_fields_in_cloud(build_config)\n else:\n # If all files can be processed by docling, do so\n allow_advanced = all(not file_path.endswith((\".csv\", \".xlsx\", \".parquet\")) for file_path in paths)\n build_config[\"advanced_mode\"][\"show\"] = allow_advanced\n if not allow_advanced:\n build_config[\"advanced_mode\"][\"value\"] = False\n docling_fields = (\n \"pipeline\",\n \"ocr_engine\",\n \"doc_key\",\n \"md_image_placeholder\",\n \"md_page_break_placeholder\",\n )\n for field in docling_fields:\n if field in build_config:\n build_config[field][\"show\"] = False\n\n # Docling Processing\n elif field_name == \"advanced_mode\":\n # Disable in cloud environments - don't show Docling fields even if advanced_mode is toggled\n if is_astra_cloud_environment():\n self._disable_docling_fields_in_cloud(build_config)\n else:\n docling_fields = (\n \"pipeline\",\n \"ocr_engine\",\n \"doc_key\",\n \"md_image_placeholder\",\n \"md_page_break_placeholder\",\n )\n for field in docling_fields:\n if field in build_config:\n build_config[field][\"show\"] = bool(field_value)\n if field == \"pipeline\":\n build_config[field][\"advanced\"] = not bool(field_value)\n\n elif field_name == \"pipeline\":\n # Disable in cloud environments - don't show OCR engine even if pipeline is changed\n if is_astra_cloud_environment():\n self._disable_docling_fields_in_cloud(build_config)\n elif field_value == \"standard\":\n build_config[\"ocr_engine\"][\"show\"] = True\n build_config[\"ocr_engine\"][\"value\"] = \"easyocr\"\n else:\n build_config[\"ocr_engine\"][\"show\"] = False\n build_config[\"ocr_engine\"][\"value\"] = \"None\"\n\n return build_config\n\n def update_outputs(self, frontend_node: dict[str, Any], field_name: str, field_value: Any) -> dict[str, Any]: # noqa: ARG002\n \"\"\"Dynamically show outputs based on file count/type and advanced mode.\"\"\"\n if field_name not in [\"path\", \"advanced_mode\", \"pipeline\"]:\n return frontend_node\n\n template = frontend_node.get(\"template\", {})\n paths = self._path_value(template)\n if not paths:\n return frontend_node\n\n frontend_node[\"outputs\"] = []\n if len(paths) == 1:\n file_path = paths[0] if field_name == \"path\" else frontend_node[\"template\"][\"path\"][\"file_path\"][0]\n if file_path.endswith((\".csv\", \".xlsx\", \".parquet\")):\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Structured Content\",\n name=\"dataframe\",\n method=\"load_files_structured\",\n types=[\"Table\"],\n selected=\"Table\",\n tool_mode=True,\n ),\n )\n elif file_path.endswith(\".json\"):\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Structured Content\",\n name=\"json\",\n method=\"load_files_json\",\n types=[\"JSON\"],\n selected=\"JSON\",\n tool_mode=True,\n ),\n )\n\n advanced_mode = frontend_node.get(\"template\", {}).get(\"advanced_mode\", {}).get(\"value\", False)\n if advanced_mode:\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Structured Output\",\n name=\"advanced_dataframe\",\n method=\"load_files_dataframe\",\n types=[\"Table\"],\n selected=\"Table\",\n tool_mode=True,\n ),\n )\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Markdown\",\n name=\"advanced_markdown\",\n method=\"load_files_markdown\",\n types=[\"Message\"],\n selected=\"Message\",\n tool_mode=True,\n ),\n )\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"File Path\",\n name=\"path\",\n method=\"load_files_path\",\n types=[\"Message\"],\n selected=\"Message\",\n tool_mode=True,\n ),\n )\n else:\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Raw Content\",\n name=\"message\",\n method=\"load_files_message\",\n types=[\"Message\"],\n selected=\"Message\",\n tool_mode=True,\n ),\n )\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"File Path\",\n name=\"path\",\n method=\"load_files_path\",\n types=[\"Message\"],\n selected=\"Message\",\n tool_mode=True,\n ),\n )\n else:\n # Multiple files => DataFrame and Message outputs; advanced parser disabled\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Files\",\n name=\"dataframe\",\n method=\"load_files\",\n types=[\"Table\"],\n selected=\"Table\",\n tool_mode=True,\n )\n )\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Raw Content\",\n name=\"message\",\n method=\"load_files_message\",\n types=[\"Message\"],\n selected=\"Message\",\n tool_mode=True,\n )\n )\n\n return frontend_node\n\n # ------------------------------ Core processing ----------------------------------\n\n def _get_selected_storage_location(self) -> str:\n \"\"\"Get the selected storage location from the SortableListInput.\"\"\"\n if hasattr(self, \"storage_location\") and self.storage_location:\n if isinstance(self.storage_location, list) and len(self.storage_location) > 0:\n return self.storage_location[0].get(\"name\", \"\")\n if isinstance(self.storage_location, dict):\n return self.storage_location.get(\"name\", \"\")\n return \"Local\" # Default to Local if not specified\n\n def _validate_and_resolve_paths(self) -> list[BaseFileComponent.BaseFile]:\n \"\"\"Override to handle file_path_str input from tool mode and cloud storage.\n\n Priority:\n 1. Cloud storage (AWS/Google Drive) if selected\n 2. file_path_str (if provided by the tool call)\n 3. path (uploaded file from UI)\n \"\"\"\n storage_location = self._get_selected_storage_location()\n\n # Handle AWS S3\n if storage_location == \"AWS\":\n return self._read_from_aws_s3()\n\n # Handle Google Drive\n if storage_location == \"Google Drive\":\n return self._read_from_google_drive()\n\n # Handle Local storage\n # Check if file_path_str is provided (from tool mode)\n file_path_str = getattr(self, \"file_path_str\", None)\n if file_path_str:\n # Use the string path from tool mode\n from pathlib import Path\n\n from lfx.schema.data import Data\n from lfx.utils.file_path_security import (\n StorageNamespaceError,\n component_file_access_scopes,\n enforce_local_file_access,\n )\n\n # Use same resolution logic as BaseFileComponent (support storage paths)\n path_str = str(file_path_str)\n if parse_storage_path(path_str):\n try:\n resolved_path = Path(self.get_full_path(path_str))\n except StorageNamespaceError:\n # A storage namespace outside this graph's scope is an access denial,\n # not a resolution failure: never retry it as a plain local path.\n raise\n except (ValueError, AttributeError):\n resolved_path = Path(self.resolve_path(path_str))\n else:\n resolved_path = Path(self.resolve_path(path_str))\n\n # Security: confine tool-mode reads to the storage dir in restricted (multi-tenant)\n # mode so a tenant cannot read arbitrary server files via file_path_str.\n resolved_path = enforce_local_file_access(resolved_path, scope_ids=component_file_access_scopes(self))\n\n if not resolved_path.exists():\n msg = f\"File or directory not found: {file_path_str}\"\n self.log(msg)\n if not self.silent_errors:\n raise ValueError(msg)\n return []\n\n data_obj = Data(data={self.SERVER_FILE_PATH_FIELDNAME: str(resolved_path)})\n return [BaseFileComponent.BaseFile(data_obj, resolved_path, delete_after_processing=False)]\n\n # Otherwise use the default implementation (uses path FileInput)\n return super()._validate_and_resolve_paths()\n\n def _read_from_aws_s3(self) -> list[BaseFileComponent.BaseFile]:\n \"\"\"Read file from AWS S3.\"\"\"\n from lfx.base.data.cloud_storage_utils import create_s3_client, validate_aws_credentials\n\n # Validate AWS credentials\n validate_aws_credentials(self)\n if not getattr(self, \"s3_file_key\", None):\n msg = \"S3 File Key is required\"\n raise ValueError(msg)\n\n # Create S3 client\n s3_client = create_s3_client(self)\n\n # Download file to temp location\n import tempfile\n\n # Get file extension from S3 key\n file_extension = Path(self.s3_file_key).suffix or \"\"\n\n temp_file_path: str | None = None\n try:\n with tempfile.NamedTemporaryFile(mode=\"wb\", suffix=file_extension, delete=False) as temp_file:\n temp_file_path = temp_file.name\n s3_client.download_fileobj(self.bucket_name, self.s3_file_key, temp_file)\n except Exception as e:\n # The context manager has already closed the handle by the time this runs, which is\n # what Windows requires before the partial download can be deleted. Keeping the\n # cleanup here also covers a failure raised by the closing flush itself.\n if temp_file_path is not None:\n with contextlib.suppress(OSError):\n Path(temp_file_path).unlink()\n msg = f\"Failed to download file from S3: {e}\"\n raise RuntimeError(msg) from e\n\n # Create BaseFile object\n from lfx.schema.data import Data\n\n temp_path = Path(temp_file_path)\n data_obj = Data(data={self.SERVER_FILE_PATH_FIELDNAME: str(temp_path)})\n return [\n BaseFileComponent.BaseFile(\n data_obj,\n temp_path,\n delete_after_processing=True,\n cleanup_local_file=True,\n )\n ]\n\n def _read_from_google_drive(self) -> list[BaseFileComponent.BaseFile]:\n \"\"\"Read file from Google Drive.\"\"\"\n import tempfile\n\n from googleapiclient.http import MediaIoBaseDownload\n\n from lfx.base.data.cloud_storage_utils import create_google_drive_service\n\n # Validate Google Drive credentials\n if not getattr(self, \"service_account_key\", None):\n msg = \"GCP Credentials Secret Key is required for Google Drive storage\"\n raise ValueError(msg)\n if not getattr(self, \"file_id\", None):\n msg = \"Google Drive File ID is required\"\n raise ValueError(msg)\n\n # Create Google Drive service with read-only scope\n drive_service = create_google_drive_service(\n self.service_account_key, scopes=[\"https://www.googleapis.com/auth/drive.readonly\"]\n )\n\n # Get file metadata to determine file name and extension\n try:\n file_metadata = drive_service.files().get(fileId=self.file_id, fields=\"name,mimeType\").execute()\n file_name = file_metadata.get(\"name\", \"download\")\n except Exception as e:\n msg = (\n f\"Unable to access file with ID '{self.file_id}'. \"\n f\"Error: {e!s}. \"\n \"Please ensure: 1) The file ID is correct, 2) The file exists, \"\n \"3) The service account has been granted access to this file.\"\n )\n raise ValueError(msg) from e\n\n # Download file to temp location\n file_extension = Path(file_name).suffix or \"\"\n temp_file_path: str | None = None\n try:\n with tempfile.NamedTemporaryFile(mode=\"wb\", suffix=file_extension, delete=False) as temp_file:\n temp_file_path = temp_file.name\n request = drive_service.files().get_media(fileId=self.file_id)\n downloader = MediaIoBaseDownload(temp_file, request)\n done = False\n while not done:\n _status, done = downloader.next_chunk()\n except Exception as e:\n # The context manager has already closed the handle by the time this runs, which is\n # what Windows requires before the partial download can be deleted. Keeping the\n # cleanup here also covers a failure raised by the closing flush itself.\n if temp_file_path is not None:\n with contextlib.suppress(OSError):\n Path(temp_file_path).unlink()\n msg = f\"Failed to download file from Google Drive: {e}\"\n raise RuntimeError(msg) from e\n\n # Create BaseFile object\n from lfx.schema.data import Data\n\n temp_path = Path(temp_file_path)\n data_obj = Data(data={self.SERVER_FILE_PATH_FIELDNAME: str(temp_path)})\n return [\n BaseFileComponent.BaseFile(\n data_obj,\n temp_path,\n delete_after_processing=True,\n cleanup_local_file=True,\n )\n ]\n\n def _is_docling_compatible(self, file_path: str) -> bool:\n \"\"\"Lightweight extension gate for Docling-compatible types.\"\"\"\n docling_exts = (\n \".adoc\",\n \".asciidoc\",\n \".asc\",\n \".bmp\",\n \".csv\",\n \".dotx\",\n \".dotm\",\n \".docm\",\n \".docx\",\n \".htm\",\n \".html\",\n \".jpg\",\n \".jpeg\",\n \".json\",\n \".md\",\n \".pdf\",\n \".png\",\n \".potx\",\n \".ppsx\",\n \".pptm\",\n \".potm\",\n \".ppsm\",\n \".pptx\",\n \".tiff\",\n \".txt\",\n \".xls\",\n \".xlsx\",\n \".xhtml\",\n \".xml\",\n \".webp\",\n )\n return file_path.lower().endswith(docling_exts)\n\n async def _get_local_file_for_docling(\n self, file_path: str, *, is_local_temp_file: bool = False\n ) -> tuple[str, bool]:\n \"\"\"Get a local file path for Docling processing, downloading from S3 if needed.\n\n Args:\n file_path: Either a local path or S3 key (format \"flow_id/filename\")\n is_local_temp_file: Whether file_path is an internally created local temporary file.\n\n Returns:\n tuple[str, bool]: (local_path, should_delete) where should_delete indicates\n if this is a temporary file that should be cleaned up\n \"\"\"\n if is_local_temp_file:\n local_temp_path = Path(file_path)\n if not local_temp_path.is_absolute() or not local_temp_path.is_file():\n msg = f\"Invalid local temporary path: {file_path}\"\n raise ValueError(msg)\n return file_path, False\n\n settings = get_settings_service().settings\n if settings.storage_type == \"local\":\n return file_path, False\n\n # S3 storage - download to temp file\n parsed = parse_storage_path(file_path)\n if not parsed:\n msg = f\"Invalid S3 path format: {file_path}. Expected 'flow_id/filename'\"\n raise ValueError(msg)\n\n storage_service = require_storage_service(get_storage_service())\n flow_id, filename = parsed\n\n # Get file content from S3\n content = await storage_service.get_file(flow_id, filename)\n\n write = asyncio.ensure_future(asyncio.to_thread(_write_temp_file, content, Path(filename).suffix))\n try:\n temp_path = await asyncio.shield(write)\n except asyncio.CancelledError:\n write.add_done_callback(_unlink_orphaned_temp_file)\n raise\n\n return temp_path, True\n\n def _process_docling_in_subprocess(self, file_path: str, *, is_local_temp_file: bool = False) -> Data | None:\n \"\"\"Sync wrapper around ``_aprocess_docling_in_subprocess``.\"\"\"\n return run_until_complete(\n self._aprocess_docling_in_subprocess(file_path, is_local_temp_file=is_local_temp_file)\n )\n\n async def _aprocess_docling_in_subprocess(self, file_path: str, *, is_local_temp_file: bool = False) -> Data | None:\n \"\"\"Run Docling in a separate OS process and map the result to a Data object.\n\n We avoid multiprocessing pickling by launching `python -c \"