diff --git a/plugins/agent_compose_strategy/CHANGELOG.md b/plugins/agent_compose_strategy/CHANGELOG.md index a5bb9ca..cfaa260 100644 --- a/plugins/agent_compose_strategy/CHANGELOG.md +++ b/plugins/agent_compose_strategy/CHANGELOG.md @@ -4,3 +4,5 @@ - Published the verified agent-compose Agent Strategy implementation in the Chaitin plugin monorepo. + +- 支持使用 Dify SDK File.blob、filename、mime_type 字段上传文件。 diff --git a/plugins/agent_compose_strategy/README.md b/plugins/agent_compose_strategy/README.md index 797f5a8..cfd9860 100644 --- a/plugins/agent_compose_strategy/README.md +++ b/plugins/agent_compose_strategy/README.md @@ -21,3 +21,7 @@ Install the versioned `.difypkg`, select **agent-compose Strategy** in an Agent With `keep_running`, the strategy remembers an agent-scoped sandbox for the Dify conversation and can reuse it on later turns. Stop and remove policies avoid retaining sandbox state. Every invocation emits text, JSON, and separate structured run variables. See [configuration and usage](docs/usage.md) and the [architecture introduction](docs/architecture.zh-CN.md). This project is licensed under Apache-2.0; see the repository-level license. + +## File uploads + +Configure workspace_id to enable uploaded files. diff --git a/plugins/agent_compose_strategy/README.zh-CN.md b/plugins/agent_compose_strategy/README.zh-CN.md index 02ed6aa..eac93cb 100644 --- a/plugins/agent_compose_strategy/README.zh-CN.md +++ b/plugins/agent_compose_strategy/README.zh-CN.md @@ -5,3 +5,7 @@ 这是一个 Dify Agent Strategy 插件，可将 Agent 节点执行委托给 [agent-compose](https://github.com/chaitin/agent-compose)，并支持隔离沙箱生命周期管理及结构化运行信息。 要求 Dify 1.15.0 或更高版本。使用 `keep_running` 时，Strategy 可以按 Dify 会话和 Agent 保存、复用沙箱；一次性任务可选择完成后停止或删除。详见[配置与使用](docs/usage.zh-CN.md)和[完整架构与选型介绍](docs/architecture.zh-CN.md)。本项目采用 Apache-2.0 许可证，以仓库根目录许可证为准。 + +## 文件上传 + +配置 workspace_id 后可启用文件上传。 diff --git a/plugins/agent_compose_strategy/client/agent_compose.py b/plugins/agent_compose_strategy/client/agent_compose.py index 4bee9af..8a922bf 100644 --- a/plugins/agent_compose_strategy/client/agent_compose.py +++ b/plugins/agent_compose_strategy/client/agent_compose.py @@ -13,6 +13,7 @@ GET_PROJECT_PROCEDURE = "/agentcompose.v2.ProjectService/GetProject" LIST_PROJECTS_PROCEDURE = "/agentcompose.v2.ProjectService/ListProjects" RUN_AGENT_PROCEDURE = "/agentcompose.v2.RunService/RunAgent" +WORKSPACE_UPLOAD_PATH = "/api/agent-compose/workspaces/{workspace_id}/upload" MAX_TIMEOUT_SECONDS = 3600 @@ -130,6 +131,7 @@ class AgentComposeAgent: driver: str = "" display_name: str = "" description: str = "" + workspace_id: str = "" def selection_value(self) -> str: return json.dumps( @@ -137,6 +139,7 @@ def selection_value(self) -> str: "project_id": self.project_id, "project_name": self.project_name, "agent_name": self.agent_name, + "workspace_id": self.workspace_id, }, ensure_ascii=False, sort_keys=True, @@ -191,6 +194,9 @@ def list_agents(self, *, page_size: int = 100) -> list[AgentComposeAgent]: agent.get("displayName") or agent.get("display_name") or "" ).strip(), description=str(agent.get("description") or "").strip(), + workspace_id=str( + agent.get("workspaceId") or agent.get("workspace_id") or "" + ).strip(), ) ) return agents @@ -228,6 +234,35 @@ def run_agent( body = self._post_json(RUN_AGENT_PROCEDURE, payload) return parse_run_agent_response(body) + def upload_workspace_file( + self, + *, + workspace_id: str, + path: str, + content: bytes, + filename: str, + content_type: str = "application/octet-stream", + ) -> None: + if not workspace_id.strip(): + raise AgentComposeError("selected agent has no file workspace configured") + url = self.config.normalized_base_url() + WORKSPACE_UPLOAD_PATH.format( + workspace_id=workspace_id.strip() + ) + headers = {"Accept": "application/json"} + if self.config.bearer_token: + headers["Authorization"] = f"Bearer {self.config.bearer_token}" + try: + response = requests.post( + url, + headers=headers, + data={"path": path, "upload_type": "file"}, + files={"file": (filename, content, content_type)}, + timeout=self.config.timeout_seconds, + ) + response.raise_for_status() + except requests.RequestException as exc: + raise AgentComposeError(f"agent-compose workspace upload failed: {exc}") from exc + def validate_connection(self) -> None: """Validate URL, authentication, and the frozen v2 Project API.""" self._post_json(LIST_PROJECTS_PROCEDURE, {"offset": 0, "limit": 1}) diff --git a/plugins/agent_compose_strategy/strategies/dynamic_workflow.py b/plugins/agent_compose_strategy/strategies/dynamic_workflow.py index 426dbed..0a6e3bd 100644 --- a/plugins/agent_compose_strategy/strategies/dynamic_workflow.py +++ b/plugins/agent_compose_strategy/strategies/dynamic_workflow.py @@ -1,10 +1,14 @@ import json +import os +import re import time +import uuid from collections.abc import Generator from typing import Any from dify_plugin.entities.agent import AgentInvokeMessage from dify_plugin.entities.tool import ToolInvokeMessage +from dify_plugin.file.file import File from dify_plugin.interfaces.agent import AgentStrategy from pydantic import BaseModel @@ -25,6 +29,8 @@ class DynamicWorkflowParams(BaseModel): agent_compose_timeout_seconds: int | None = None agent: str query: str + files: list[File] | None = None + workspace_id: str | None = None instruction: str | None = None cleanup_policy: str = "stop_on_completion" output_schema_json: str | None = None @@ -34,7 +40,6 @@ class DynamicWorkflowParams(BaseModel): class DynamicWorkflowAgentStrategy(AgentStrategy): def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage, None, None]: params = DynamicWorkflowParams(**parameters) - prompt = build_prompt(params.instruction, params.query) client = AgentComposeClient( AgentComposeConfig.from_mapping( { @@ -44,6 +49,8 @@ def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage, N } ) ) + file_paths = upload_files(client, params.workspace_id or "", params.files, self.session) + prompt = build_prompt(params.instruction, params.query, file_paths) project_id, agent_name = resolve_agent_reference(client, params.agent) reuse_sandbox = cleanup_policy_reuses_sandbox(params.cleanup_policy) sandbox_id = "" @@ -140,15 +147,49 @@ def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage, N raise AgentComposeError(failure_reason) -def build_prompt(instruction: str | None, query: str) -> str: +def build_prompt(instruction: str | None, query: str, file_paths: list[str] | None = None) -> str: instruction = (instruction or "").strip() query = query.strip() - if not instruction: + if not instruction and not file_paths: return query - return json.dumps( - { - "instruction": instruction, - "query": query, - }, - ensure_ascii=False, + payload = {"instruction": instruction, "query": query} + if file_paths: + payload["files"] = file_paths + return json.dumps(payload, ensure_ascii=False) + + +def upload_files(client, workspace_id: str, files, session) -> list[str]: + if not files: + return [] + if not workspace_id: + raise AgentComposeError("workspace_id is required when files are provided") + request_id = ( + re.sub(r"[^A-Za-z0-9_-]", "", str(getattr(session, "conversation_id", "") or "")) + or uuid.uuid4().hex ) + paths = [] + total = 0 + for index, item in enumerate(files): + data = item if isinstance(item, dict) else getattr(item, "__dict__", {}) + name = ( + os.path.basename(str(data.get("filename") or data.get("name") or f"file-{index}")) + or f"file-{index}" + ) + content = data.get("content") or getattr(item, "blob", None) + if isinstance(content, str): + content = content.encode() + if not isinstance(content, (bytes, bytearray)): + raise AgentComposeError(f"unable to read uploaded file {name}") + if len(content) > 50 * 1024 * 1024 or total + len(content) > 100 * 1024 * 1024: + raise AgentComposeError("uploaded files exceed size limits") + total += len(content) + path = f"inputs/{request_id}/{index}-{name}" + client.upload_workspace_file( + workspace_id=workspace_id, + path=path, + content=bytes(content), + filename=name, + content_type=str(data.get("mime_type") or "application/octet-stream"), + ) + paths.append(path) + return paths diff --git a/plugins/agent_compose_strategy/strategies/dynamic_workflow.yaml b/plugins/agent_compose_strategy/strategies/dynamic_workflow.yaml index 2df3f68..7ea2a0f 100644 --- a/plugins/agent_compose_strategy/strategies/dynamic_workflow.yaml +++ b/plugins/agent_compose_strategy/strategies/dynamic_workflow.yaml @@ -56,6 +56,18 @@ parameters: label: en_US: Query zh_Hans: 查询 + - name: files + type: files + required: false + label: + en_US: Files + zh_Hans: 文件 + - name: workspace_id + type: string + required: false + label: + en_US: Workspace ID + zh_Hans: 工作区 ID - name: cleanup_policy type: string required: false diff --git a/plugins/agent_compose_strategy/tests/test_file_upload.py b/plugins/agent_compose_strategy/tests/test_file_upload.py new file mode 100644 index 0000000..f473fa7 --- /dev/null +++ b/plugins/agent_compose_strategy/tests/test_file_upload.py @@ -0,0 +1,43 @@ +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from dify_plugin.file.entities import FileType +from dify_plugin.file.file import File + +from client.agent_compose import AgentComposeError +from strategies.dynamic_workflow import build_prompt, upload_files + + +def test_upload_files_reads_dify_file_blob_and_isolates_path() -> None: + file = File( + url="https://files.example/report.pdf", filename="../report.pdf", type=FileType.DOCUMENT + ) + file._blob = b"pdf" + client = Mock() + + paths = upload_files(client, "workspace-1", [file], SimpleNamespace(conversation_id="conv/1")) + + assert paths == ["inputs/conv1/0-report.pdf"] + client.upload_workspace_file.assert_called_once_with( + workspace_id="workspace-1", + path=paths[0], + content=b"pdf", + filename="report.pdf", + content_type="application/octet-stream", + ) + assert '"files": ["inputs/conv1/0-report.pdf"]' in build_prompt("", "question", paths) + + +def test_upload_files_validates_workspace_and_size() -> None: + file = File(url="https://files.example/large.pdf", filename="large.pdf", type=FileType.DOCUMENT) + file._blob = b"x" * (50 * 1024 * 1024 + 1) + with pytest.raises(AgentComposeError, match="workspace_id"): + upload_files(Mock(), "", [file], SimpleNamespace()) + with pytest.raises(AgentComposeError, match="size limits"): + upload_files(Mock(), "workspace-1", [file], SimpleNamespace()) + + +def test_upload_files_empty_is_backward_compatible() -> None: + assert upload_files(Mock(), "", None, SimpleNamespace()) == [] + assert build_prompt(None, "question", []) == "question" diff --git a/plugins/agent_compose_workflow/CHANGELOG.md b/plugins/agent_compose_workflow/CHANGELOG.md index 5d40540..b65ef91 100644 --- a/plugins/agent_compose_workflow/CHANGELOG.md +++ b/plugins/agent_compose_workflow/CHANGELOG.md @@ -4,3 +4,5 @@ - Published the verified agent-compose Workflow Tool implementation as `agent_compose_workflow` in the Chaitin plugin monorepo. + +- Files use Dify SDK File.blob, filename, and mime_type fields. diff --git a/plugins/agent_compose_workflow/README.md b/plugins/agent_compose_workflow/README.md index 5faf3ee..c02fda2 100644 --- a/plugins/agent_compose_workflow/README.md +++ b/plugins/agent_compose_workflow/README.md @@ -23,3 +23,7 @@ The tool accepts an agent, query, optional instruction, cleanup policy, optional Choose `stop_on_completion` for ordinary one-shot work, `keep_running` when a later request should reuse the sandbox, or `remove_on_completion` when the workspace must be discarded. See [configuration and usage](docs/usage.md) and the [architecture introduction](docs/architecture.zh-CN.md). This project is licensed under Apache-2.0; see the repository-level license. + +## File uploads + +Both integrations accept optional files and upload them to a configured agent-compose workspace under a conversation-scoped inputs directory. Configure workspace_id when agent metadata does not provide it. diff --git a/plugins/agent_compose_workflow/README.zh-CN.md b/plugins/agent_compose_workflow/README.zh-CN.md index c0f7c75..62153a4 100644 --- a/plugins/agent_compose_workflow/README.zh-CN.md +++ b/plugins/agent_compose_workflow/README.zh-CN.md @@ -7,3 +7,7 @@ 要求 Dify 1.15.0 或更高版本。配置 agent-compose 基础地址和可选 Bearer Token 后，在工作流中加入 **运行动态工作流** 工具即可。 详见[配置与使用](docs/usage.zh-CN.md)和[完整架构与选型介绍](docs/architecture.zh-CN.md)。插件支持动态选择 Agent、沙箱清理策略、结构化输出和幂等请求 ID，并返回文本及结构化运行信息。本项目采用 Apache-2.0 许可证，以仓库根目录许可证为准。 + +## 文件上传 + +两个集成都支持可选 files 参数，文件会上传到 agent-compose 工作区的会话隔离 inputs 目录。若智能体元数据未提供工作区 ID，请配置 workspace_id。 diff --git a/plugins/agent_compose_workflow/client/agent_compose.py b/plugins/agent_compose_workflow/client/agent_compose.py index 4bee9af..8a922bf 100644 --- a/plugins/agent_compose_workflow/client/agent_compose.py +++ b/plugins/agent_compose_workflow/client/agent_compose.py @@ -13,6 +13,7 @@ GET_PROJECT_PROCEDURE = "/agentcompose.v2.ProjectService/GetProject" LIST_PROJECTS_PROCEDURE = "/agentcompose.v2.ProjectService/ListProjects" RUN_AGENT_PROCEDURE = "/agentcompose.v2.RunService/RunAgent" +WORKSPACE_UPLOAD_PATH = "/api/agent-compose/workspaces/{workspace_id}/upload" MAX_TIMEOUT_SECONDS = 3600 @@ -130,6 +131,7 @@ class AgentComposeAgent: driver: str = "" display_name: str = "" description: str = "" + workspace_id: str = "" def selection_value(self) -> str: return json.dumps( @@ -137,6 +139,7 @@ def selection_value(self) -> str: "project_id": self.project_id, "project_name": self.project_name, "agent_name": self.agent_name, + "workspace_id": self.workspace_id, }, ensure_ascii=False, sort_keys=True, @@ -191,6 +194,9 @@ def list_agents(self, *, page_size: int = 100) -> list[AgentComposeAgent]: agent.get("displayName") or agent.get("display_name") or "" ).strip(), description=str(agent.get("description") or "").strip(), + workspace_id=str( + agent.get("workspaceId") or agent.get("workspace_id") or "" + ).strip(), ) ) return agents @@ -228,6 +234,35 @@ def run_agent( body = self._post_json(RUN_AGENT_PROCEDURE, payload) return parse_run_agent_response(body) + def upload_workspace_file( + self, + *, + workspace_id: str, + path: str, + content: bytes, + filename: str, + content_type: str = "application/octet-stream", + ) -> None: + if not workspace_id.strip(): + raise AgentComposeError("selected agent has no file workspace configured") + url = self.config.normalized_base_url() + WORKSPACE_UPLOAD_PATH.format( + workspace_id=workspace_id.strip() + ) + headers = {"Accept": "application/json"} + if self.config.bearer_token: + headers["Authorization"] = f"Bearer {self.config.bearer_token}" + try: + response = requests.post( + url, + headers=headers, + data={"path": path, "upload_type": "file"}, + files={"file": (filename, content, content_type)}, + timeout=self.config.timeout_seconds, + ) + response.raise_for_status() + except requests.RequestException as exc: + raise AgentComposeError(f"agent-compose workspace upload failed: {exc}") from exc + def validate_connection(self) -> None: """Validate URL, authentication, and the frozen v2 Project API.""" self._post_json(LIST_PROJECTS_PROCEDURE, {"offset": 0, "limit": 1}) diff --git a/plugins/agent_compose_workflow/tests/test_file_upload.py b/plugins/agent_compose_workflow/tests/test_file_upload.py new file mode 100644 index 0000000..7f244f7 --- /dev/null +++ b/plugins/agent_compose_workflow/tests/test_file_upload.py @@ -0,0 +1,42 @@ +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from dify_plugin.file.entities import FileType +from dify_plugin.file.file import File + +from client.agent_compose import AgentComposeError +from tools.run_agent import build_prompt, upload_files + + +def test_upload_files_reads_dify_file_blob_and_isolates_path() -> None: + file = File( + url="https://files.example/report.pdf", filename="../report.pdf", type=FileType.DOCUMENT + ) + file._blob = b"pdf" + client = Mock() + + paths = upload_files(client, "workspace-1", [file], SimpleNamespace(conversation_id="conv/1")) + + assert paths == ["inputs/conv1/0-report.pdf"] + client.upload_workspace_file.assert_called_once_with( + workspace_id="workspace-1", + path=paths[0], + content=b"pdf", + filename="report.pdf", + content_type="application/octet-stream", + ) + assert '"files": ["inputs/conv1/0-report.pdf"]' in build_prompt("", "question", paths) + + +def test_upload_files_validates_workspace_and_size() -> None: + file = {"filename": "large.pdf", "content": b"x" * (50 * 1024 * 1024 + 1)} + with pytest.raises(AgentComposeError, match="workspace_id"): + upload_files(Mock(), "", [file], SimpleNamespace()) + with pytest.raises(AgentComposeError, match="size limits"): + upload_files(Mock(), "workspace-1", [file], SimpleNamespace()) + + +def test_upload_files_empty_is_backward_compatible() -> None: + assert upload_files(Mock(), "", None, SimpleNamespace()) == [] + assert build_prompt(None, "question", []) == "question" diff --git a/plugins/agent_compose_workflow/tools/run_agent.py b/plugins/agent_compose_workflow/tools/run_agent.py index 0e7318b..cd728a2 100644 --- a/plugins/agent_compose_workflow/tools/run_agent.py +++ b/plugins/agent_compose_workflow/tools/run_agent.py @@ -1,5 +1,8 @@ import json +import os +import re import time +import uuid from collections.abc import Generator from typing import Any @@ -24,6 +27,7 @@ def _invoke( tool_parameters: dict[str, Any], ) -> Generator[ToolInvokeMessage, None, None]: project_id, agent_name = parse_agent_selection(str(tool_parameters.get("agent") or "")) + selection = json.loads(str(tool_parameters.get("agent") or "{}")) query = str(tool_parameters.get("query") or "").strip() instruction = str(tool_parameters.get("instruction") or "").strip() if not query: @@ -53,13 +57,19 @@ def _invoke( ) yield run_log + client = AgentComposeClient(AgentComposeConfig.from_mapping(self.runtime.credentials)) + file_paths = upload_files( + client, + str(tool_parameters.get("workspace_id") or selection.get("workspace_id", "")), + tool_parameters.get("files"), + self.session, + ) + prompt = build_prompt(instruction, query, file_paths) try: - result = AgentComposeClient( - AgentComposeConfig.from_mapping(self.runtime.credentials) - ).run_agent( + result = client.run_agent( project_id=project_id, agent_name=agent_name, - prompt=build_prompt(instruction, query), + prompt=prompt, sandbox_id=sandbox_id, cleanup_policy=cleanup_policy, output_schema_json=str(tool_parameters.get("output_schema_json") or ""), @@ -153,12 +163,56 @@ def agent_option_label(project_name: str, agent_name: str, display_name: str = " return f"{project_name}/{agent_label}" if project_name else agent_label -def build_prompt(instruction: str | None, query: str) -> str: +def build_prompt(instruction: str | None, query: str, file_paths: list[str] | None = None) -> str: instruction = (instruction or "").strip() query = query.strip() - if not instruction: + if not instruction and not file_paths: return query + payload = {"instruction": instruction, "query": query} + if file_paths: + payload["files"] = file_paths return json.dumps( - {"instruction": instruction, "query": query}, + payload, ensure_ascii=False, ) + + +def upload_files(client, workspace_id: str, files, session) -> list[str]: + if not files: + return [] + if not workspace_id: + raise AgentComposeError("workspace_id is required when files are provided") + if isinstance(files, dict): + files = [files] + request_id = ( + re.sub(r"[^A-Za-z0-9_-]", "", str(getattr(session, "conversation_id", "") or "")) + or uuid.uuid4().hex + ) + paths = [] + total = 0 + for index, item in enumerate(files): + data = item if isinstance(item, dict) else getattr(item, "__dict__", {}) + name = ( + os.path.basename(str(data.get("filename") or data.get("name") or f"file-{index}")) + or f"file-{index}" + ) + content = data.get("content") or getattr(item, "blob", None) + if isinstance(content, str): + content = content.encode() + if not isinstance(content, (bytes, bytearray)): + raise AgentComposeError(f"unable to read uploaded file {name}") + if len(content) > 50 * 1024 * 1024 or total + len(content) > 100 * 1024 * 1024: + raise AgentComposeError("uploaded files exceed size limits") + total += len(content) + path = f"inputs/{request_id}/{index}-{name}" + client.upload_workspace_file( + workspace_id=workspace_id, + path=path, + content=bytes(content), + filename=name, + content_type=str( + data.get("mime_type") or data.get("mimeType") or "application/octet-stream" + ), + ) + paths.append(path) + return paths diff --git a/plugins/agent_compose_workflow/tools/run_agent.yaml b/plugins/agent_compose_workflow/tools/run_agent.yaml index 2880a7e..b2f16c4 100644 --- a/plugins/agent_compose_workflow/tools/run_agent.yaml +++ b/plugins/agent_compose_workflow/tools/run_agent.yaml @@ -32,6 +32,24 @@ parameters: en_US: User query passed to the selected agent. zh_Hans: 传递给所选智能体的用户查询。 llm_description: User query passed to the selected agent. + - name: files + type: files + form: llm + required: false + label: + en_US: Files + zh_Hans: 文件 + human_description: + en_US: Optional files uploaded to the selected agent workspace. + zh_Hans: 上传到所选智能体工作区的可选文件。 + llm_description: Files for the agent to inspect. + - name: workspace_id + type: string + form: form + required: false + label: + en_US: Workspace ID + zh_Hans: 工作区 ID - name: instruction type: string form: form