Self Checks

• I have read the Contributing Guide and Language Policy.
• This is only for bug report, if you would like to ask a question, please head to Discussions.
• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report, otherwise it will be closed.
• 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
• Please do not modify this template :) and fill in all the required fields.

Dify version

1.4.3

Cloud or Self Hosted

Self Hosted (Docker)

Steps to reproduce

The plugin I developed is designed to concurrently retrieve knowledge from Dify's knowledge base. During debugging, the plugin works normally, but after being packaged and uploaded to Dify, it fails to connect to any network whatsoever. Requests can only be sent successfully when routed through a proxy. What could be the reason for this? In the official example code, I noticed that no proxy settings are required.

import ast
import asyncio
import json
import logging
import os
from collections.abc import Generator
from typing import Any, Optional, Dict

import httpx
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage

import gevent
import gevent.monkey
gevent.monkey.patch_all()

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class KnowledgeQueryTool(Tool):
    async def query_async(self, client: httpx.AsyncClient, question: str,
                          DIFY_URL=None, DATASET_ID=None, BEARER_TOKEN=None) -> Optional[Dict[str, Any]]:
        """使用 httpx.AsyncClient 异步查询 Dify API。"""
        URL = f"{DIFY_URL}/datasets/{DATASET_ID}/retrieve"
        HEADERS = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {BEARER_TOKEN}",
        }
        payload = {
            "query": question,
            "retrieval_model": {
                "search_mode": "semantic_search",
                "reranking_enable": False,
                "top_k": 10,
                "score_threshold_enabled": True,
                "score_threshold": 0.27
            }
        }
        try:
            logger.debug(f"发送异步请求: URL={URL}, question={question}, payload={payload}")
            response = await client.post(URL, headers=HEADERS, json=payload, timeout=60)
            logger.debug(f"响应状态码: {response.status_code}, 内容: {response.text}")
            response.raise_for_status()
            return response.json()
        except httpx.RequestError as e:
            logger.error(f"异步请求失败: {e}")
            return None

    async def main(self, dify_url, dify_app_key, dify_database_id, questions_str):
        logger.info(f"main 函数启动: questions_str={questions_str}")
        try:
            data_list = ast.literal_eval(questions_str)
            if not isinstance(data_list, list):
                raise TypeError("Query parameter is not a list after evaluation.")
            logger.debug(f"解析问题列表: {data_list}")
        except (ValueError, SyntaxError, TypeError) as e:
            logger.error(f"无法解析问题列表: {questions_str}. 错误: {e}")
            return [f"Error: Invalid query format. Got: {questions_str}"]

        # 代理逻辑：如果 DIFY_URL 是外部，添加 SSRF Proxy
        mounts = None
        if "localhost" not in dify_url and "api" not in dify_url.lower():  # 假设内部 URL 包含这些
            proxy_url = "http://ssrf_proxy:3128"
            mounts = {
                "http://": httpx.AsyncHTTPTransport(proxy=proxy_url),
                "https://": httpx.AsyncHTTPTransport(proxy=proxy_url)
            }
            logger.debug("DIFY_URL 是外部，使用 SSRF Proxy")

        async with httpx.AsyncClient(mounts=mounts) as client:  # 使用 mounts 配置代理
            tasks = [self.query_async(client, q, dify_url, dify_database_id, dify_app_key) for q in data_list]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        contents = []
        for question, result in zip(data_list, results):
            if isinstance(result, Exception):
                logger.error(f"问题 '{question}' 的异步请求失败: {result}")
                contents.append(f"Error processing question '{question}': {str(result)}")
            elif result is not None and 'records' in result:
                records = result.get('records', [])
                if records:
                    for data in records:
                        contents.append(data['segment']['content'])
                    logger.debug(f"问题 '{question}' 处理成功: {len(records)} 条记录")
                else:
                    logger.warning(f"问题 '{question}' 返回空记录，完整响应: {result}")
                    contents.append(f"No records found for '{question}'. Full response: {json.dumps(result)}")
            else:
                logger.warning(f"问题 '{question}' 没有返回有效记录（result={result}）")
                contents.append(f"No valid records for '{question}'. Response: {json.dumps(result) if result else 'None'}")

        logger.info(f"main 函数结束: contents={contents}")
        return contents

    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
        questions_str = tool_parameters["query"]
        dify_url = tool_parameters["dify_api_url"]
        dify_app_key = tool_parameters["dify_knowledge_key"]
        dify_database_id = tool_parameters["dify_knowledge_id"]

        logger.info(f"_invoke 调用: parameters={tool_parameters}")
        contents = asyncio.run(self.main(dify_url, dify_app_key, dify_database_id, questions_str))

        if not contents:
            logger.warning("contents 为空，返回默认消息")
            yield self.create_json_message({
                "result": "No content returned. Check if dataset or query is valid."
            })
        else:
            yield self.create_json_message({
                "result": str(contents),
            })
        logger.info("_invoke 结束")

✔️ Expected Behavior

plugin runs successfully without using a proxy.

❌ Actual Behavior

plugin can not send requests without using a proxy.