@@ -506,9 +506,9 @@ def _handle_mcp(
 
         # 工具(workflow/custom) + 智能体(子应用) → LangChain tools；
         # MCP(自定义/库内) → mcp_servers 配置；技能 → 交给引擎侧 init_skills 初始化
-        tools = get_tool_tools(source_type, source_id, tool_ids, workspace_id) + get_application_tools(
-            source_type, source_id, application_ids, workspace_id, self.get_workflow_parameters()
-        )
+        tools = get_tool_tools(
+            source_type, source_id, tool_ids, workspace_id, self.get_workflow_parameters()
+        ) + get_application_tools(source_type, source_id, application_ids, workspace_id, self.get_workflow_parameters())
         mcp_servers_config = get_mcp_servers(mcp_source, mcp_servers, mcp_tool_id, mcp_tool_ids, self._handle_variables)
         ToolExecutor().validate_mcp_transport(json.dumps(mcp_servers_config))
 
 
@@ -13,14 +13,14 @@
 __all__ = ["get_tool_tools", "get_workflow_tools", "get_custom_tools"]
 
 
-def get_tool_tools(source_type, source_id, tool_ids, workspace_id):
+def get_tool_tools(source_type, source_id, tool_ids, workspace_id, workflow_params=None):
     """
     构建工具（Tool）类工具：内部按 tool_type 拆分 workflow / custom，合并返回 LangChain tools。
 
     节点只需传入混合的 tool_ids，各构建器各自按 tool_type 过滤。
     """
     if not tool_ids:
         return []
-    return get_workflow_tools(source_type, source_id, tool_ids, workspace_id) + get_custom_tools(
+    return get_workflow_tools(source_type, source_id, tool_ids, workspace_id, workflow_params) + get_custom_tools(
         source_type, source_id, tool_ids, workspace_id
     )
@@ -81,7 +81,9 @@ def _save_workflow_tool_record(
     ).save()
 
 
-def get_workflow_func(source_type, source_id, tool, qv, workspace_id):
+def get_workflow_func(source_type, source_id, tool, qv, workspace_id, workflow_params=None):
+    from knowledge.services.retrieval_access import inherited_retrieval_context
+
     tool_id = tool.id
 
     def inner(**kwargs):
@@ -100,6 +102,7 @@ def inner(**kwargs):
             "workspace_id": workspace_id,
             "default_model_setting": qv.default_model_setting or {},
             **kwargs,
+            **inherited_retrieval_context({"workspace_id": workspace_id, **(workflow_params or {})}),
         }
 
         # WorkflowManage.run() 在后台线程异步执行节点，完成时机由 on_complete 回调驱动，
@@ -157,7 +160,7 @@ def get_start_node_fn(wf, wm):
     return inner
 
 
-def get_workflow_tools(source_type, source_id, tool_workflow_ids, workspace_id):
+def get_workflow_tools(source_type, source_id, tool_workflow_ids, workspace_id, workflow_params=None):
     tools = QuerySet(Tool).filter(
         id__in=tool_workflow_ids, is_active=True, tool_type=ToolType.WORKFLOW, workspace_id=workspace_id
     )
@@ -170,7 +173,7 @@ def get_workflow_tools(source_type, source_id, tool_workflow_ids, workspace_id):
     results = []
     for tool in tools:
         qv = qd.get(tool.id)
-        func = get_workflow_func(source_type, source_id, tool, qv, workspace_id)
+        func = get_workflow_func(source_type, source_id, tool, qv, workspace_id, workflow_params)
         args = get_workflow_args(tool, qv)
         tool = StructuredTool.from_function(
             func=func,
 
@@ -13,10 +13,9 @@
 
 from application.workflow.common import WorkflowType
 from application.workflow.i_node import INode
-from common.auth.constants.role_constants import RoleConstants
-from common.database_model_manage.database_model_manage import DatabaseModelManage
 from common.utils.shared_resource_auth import filter_authorized_ids
 from knowledge.models import Document, DocumentTag, Knowledge
+from knowledge.services.retrieval_access import filter_workflow_knowledge
 
 
 class SearchDocumentNodeSerializer(serializers.Serializer):
@@ -213,21 +212,20 @@ def execute(self):
                     QuerySet(Document).filter(knowledge_id__in=ref_knowledge_ids).values_list("id", flat=True)
                 )
 
-        get_knowledge_list_of_authorized = DatabaseModelManage.get_model("get_knowledge_list_of_authorized")
-        chat_user_type = workflow_params.get("chat_user_type")
-
-        if get_knowledge_list_of_authorized is not None and RoleConstants.CHAT_USER.value.name == chat_user_type:
-            actual_knowledge_ids = list(
-                QuerySet(Document).filter(id__in=document_id_list).values_list("knowledge_id", flat=True).distinct()
-            )
-            authorized_knowledge_ids = get_knowledge_list_of_authorized(
-                workflow_params.get("chat_user_id"), [str(k_id) for k_id in actual_knowledge_ids]
-            )
-            document_id_list = list(
-                QuerySet(Document)
-                .filter(id__in=document_id_list, knowledge_id__in=authorized_knowledge_ids)
-                .values_list("id", flat=True)
-            )
+        actual_knowledge_ids = list(
+            QuerySet(Document)
+            .filter(id__in=document_id_list or [], is_active=True)
+            .values_list("knowledge_id", flat=True)
+            .distinct()
+        )
+        authorized_knowledge_ids = filter_workflow_knowledge(
+            filter_authorized_ids("knowledge", actual_knowledge_ids, workspace_id), workflow_params
+        )
+        document_id_list = list(
+            QuerySet(Document)
+            .filter(id__in=document_id_list or [], knowledge_id__in=authorized_knowledge_ids, is_active=True)
+            .values_list("id", flat=True)
+        )
 
         if search_mode == "auto":
             matched_doc_ids = _handle_auto_tags(self.workflow_manage, document_id_list, question_reference)
@@ -239,7 +237,13 @@ def execute(self):
             final_document_ids = list(matched_document_ids)
 
         final_document_ids = [str(doc_id) for doc_id in final_document_ids]
-        document_items = list(QuerySet(Document).filter(id__in=final_document_ids).values())
+        authorized_knowledge_ids = filter_workflow_knowledge(authorized_knowledge_ids, workflow_params)
+        document_items = list(
+            QuerySet(Document)
+            .filter(id__in=final_document_ids, knowledge_id__in=authorized_knowledge_ids, is_active=True)
+            .values()
+        )
+        final_document_ids = [str(doc["id"]) for doc in document_items]
         final_knowledge_ids = list(set(str(doc["knowledge_id"]) for doc in document_items))
         knowledge_items = list(QuerySet(Knowledge).filter(id__in=final_knowledge_ids).values())
 
 
@@ -19,15 +19,14 @@
 
 from application.workflow.common import WorkflowType
 from application.workflow.i_node import INode
-from common.auth.constants.role_constants import RoleConstants
 from common.config.embedding_config import VectorStore
-from common.database_model_manage.database_model_manage import DatabaseModelManage
 from common.db.search import native_search
 from common.utils.common import flat_map, get_file_content
 from common.utils.shared_resource_auth import filter_authorized_ids
 from knowledge.models import Document, Paragraph, Knowledge, SearchMode, SourceType
 from knowledge.services.multimodal_retrieval import get_hit_asset_map
 from knowledge.services.retrieval_stats import get_recall_tracker, record_recall_safely
+from knowledge.services.retrieval_access import filter_workflow_knowledge
 from maxkb.conf import PROJECT_DIR
 from models_provider.tools import get_model_instance_by_model_workspace_id
 
@@ -178,18 +177,27 @@ def _record_recalled_items(embedding_list: List[Dict], paragraph_list: List[Dict
     record_recall_safely(recalled_items, tracker=get_recall_tracker(workflow_manage))
 
 
-def _list_paragraph(embedding_list: List, vector):
+def _list_paragraph(embedding_list: List, vector, knowledge_ids=None, document_ids=None):
     paragraph_id_list = [row.get("paragraph_id") for row in embedding_list]
     if paragraph_id_list is None or len(paragraph_id_list) == 0:
         return []
+    query = QuerySet(Paragraph).filter(id__in=paragraph_id_list)
+    if knowledge_ids is not None:
+        query = query.filter(
+            knowledge_id__in=knowledge_ids,
+            is_active=True,
+            document_id__in=QuerySet(Document).filter(knowledge_id__in=knowledge_ids, is_active=True).values("id"),
+        )
+    if document_ids is not None:
+        query = query.filter(document_id__in=document_ids)
     paragraph_list = native_search(
-        QuerySet(Paragraph).filter(id__in=paragraph_id_list),
+        query,
         get_file_content(
             os.path.join(PROJECT_DIR, "apps", "application", "sql", "list_knowledge_paragraph_by_paragraph_id.sql")
         ),
         with_table_name=True,
     )
-    if len(paragraph_list) != len(paragraph_id_list):
+    if knowledge_ids is None and len(paragraph_list) != len(paragraph_id_list):
         exist_paragraph_list = [row.get("id") for row in paragraph_list]
         for paragraph_id in paragraph_id_list:
             if paragraph_id not in exist_paragraph_list:
@@ -250,15 +258,11 @@ def execute(self):
                     .distinct()
                 ]
 
-        get_knowledge_list_of_authorized = DatabaseModelManage.get_model("get_knowledge_list_of_authorized")
-        chat_user_type = workflow_params.get("chat_user_type")
-        if get_knowledge_list_of_authorized is not None and RoleConstants.CHAT_USER.value.name == chat_user_type:
-            knowledge_id_list = get_knowledge_list_of_authorized(workflow_params.get("chat_user_id"), knowledge_id_list)
-
         workspace_id = workflow_params.get("workspace_id")
         knowledge_id_list = filter_authorized_ids("knowledge", knowledge_id_list, workspace_id)
+        knowledge_id_list = filter_workflow_knowledge(knowledge_id_list, workflow_params)
 
-        if len(knowledge_id_list) == 0:
+        if len(knowledge_id_list) == 0 or document_id_list == []:
             self._write_empty_result(question)
             return
 
@@ -293,7 +297,8 @@ def execute(self):
             self._write_empty_result(question)
             return
 
-        paragraph_list = _list_paragraph(embedding_list, vector)
+        knowledge_id_list = filter_workflow_knowledge(knowledge_id_list, workflow_params)
+        paragraph_list = _list_paragraph(embedding_list, vector, knowledge_id_list, document_id_list)
         hit_asset_map = get_hit_asset_map(embedding_list)
         result = [
             reset_paragraph
 
@@ -22,6 +22,7 @@
 from common.exception.app_exception import ChatException, AppApiException
 from common.field.common import ObjectField
 from tools.models import Tool, ToolType, ToolWorkflowVersion
+from knowledge.services.retrieval_access import inherited_retrieval_context
 
 
 class InputField(serializers.Serializer):
@@ -108,6 +109,7 @@ def execute(self):
             "form_data": workflow_params.get("form_data"),
             "default_model_setting": tool_workflow_version.default_model_setting or {},
             **parameters,
+            **inherited_retrieval_context(workflow_params),
         }
 
         node_id = self.get_node_id()
 
@@ -0,0 +1,66 @@
+"""Knowledge equivalent of the application's MCPToolHandler."""
+
+import json
+
+from rest_framework.exceptions import ValidationError
+
+from knowledge.services.external_retrieval import retrieve
+from knowledge.services.retrieval_access import RetrievalError
+
+PROTOCOL_VERSIONS = ("2025-03-26", "2025-06-18", "2025-11-25")
+
+
+class KnowledgeMCPToolHandler:
+    def __init__(self, knowledge, identity):
+        self.knowledge, self.identity = knowledge, identity
+        self.tool_name = f"knowledge_{knowledge.id}"
+
+    def initialize(self, params):
+        version = params.get("protocolVersion")
+        if (
+            not isinstance(version, str)
+            or not isinstance(params.get("capabilities"), dict)
+            or not isinstance(params.get("clientInfo"), dict)
+        ):
+            raise ValidationError("Invalid initialization parameters.")
+        return {
+            "protocolVersion": version if version in PROTOCOL_VERSIONS else "2025-06-18",
+            "serverInfo": {"name": "maxkb-knowledge-mcp", "version": "1.0.0"},
+            "capabilities": {"tools": {}},
+        }
+
+    def list_tools(self):
+        return {
+            "tools": [
+                {
+                    "name": self.tool_name,
+                    "description": f"检索知识库：{self.knowledge.name}",
+                    "inputSchema": {
+                        "type": "object",
+                        "additionalProperties": False,
+                        "required": ["query_text"],
+                        "properties": {
+                            "query_text": {"type": "string", "minLength": 1, "maxLength": 8000},
+                            "top_number": {"type": "integer", "minimum": 1, "maximum": 50, "default": 5},
+                            "similarity": {"type": "number", "minimum": 0, "maximum": 1, "default": 0},
+                            "search_mode": {
+                                "type": "string",
+                                "enum": ["embedding", "keywords", "blend"],
+                                "default": "embedding",
+                            },
+                        },
+                    },
+                }
+            ]
+        }
+
+    def call_tool(self, params):
+        if params.get("name") != self.tool_name:
+            raise ValidationError("Unknown tool.")
+        try:
+            output = retrieve(self.knowledge.id, self.identity, params.get("arguments", {}))
+        except (RetrievalError, ValidationError):
+            raise
+        except Exception:
+            return {"isError": True, "content": [{"type": "text", "text": "Knowledge retrieval failed."}]}
+        return {"content": [{"type": "text", "text": json.dumps(output, ensure_ascii=False)}]}
@@ -45,6 +45,7 @@
 from application.workflow.nodes import get_start_node
 from application.workflow.workflow_manage import WorkflowManage, CallBack
 from application.workflow.workflow_run_registry import WorkflowRunRegistry
+from knowledge.services.retrieval_access import identity_from_server
 from chat.template.agent_simple import build_workflow
 from common import result
 from common.exception.app_exception import AppApiException, AppChatNumOutOfBoundsFailed, ChatException
@@ -285,6 +286,7 @@ def chat_work_flow(self, application, instance: dict, base_to_response):
             ChatRecord(id=chat_record_id_str, **self.get_defaults_record(message_dict)).save(force_insert=True)
 
         parameters = {
+            "retrieval_identity": identity_from_server(chat_user_id, chat_user_type, debug),
             "history_chat_record": history_chat_record,
             "question": message,
             "chat_id": chat_id,
 
@@ -2,12 +2,15 @@
 
 from application.views import ChatRecordDetailView, ChatRecordLinkView
 from chat.views import v2 as v2_views, v3 as v3_views
+from chat.views.v3.knowledge import knowledge_mcp_view, retrieve_view
 
 app_name = "chat"
 # @formatter:off
 # fmt: off
 
 v3=[
+ path('knowledge/<uuid:knowledge_id>/retrieve', retrieve_view),
+ path('knowledge/<uuid:knowledge_id>/mcp', knowledge_mcp_view),
  # ---- application 作用域：application_id 从 path 获取 ----
  path('application/<str:application_id>/',  include([
   path("profile",v3_views.ApplicationProfile.as_view(), name='v3_profile'),