feat(mcp): refine English descriptions and keep utility tools agent-only

This commit is contained in:
shiyu committed 2026-10-04 11:19:43 +08:00
1 parent e897f3de87
commit 404ea4035f
10 files changed
+231 -108

No files matched your search

+2 -1
View File
@@ -81,7 +81,8 @@ An integrated AI agent with built-in tools for VFS operations, web fetching, and
- The MCP endpoint is `/api/mcp/`. External clients authenticate with a Foxel Bearer Token and own their confirmation UI; the server always enforces the authenticated user's path permissions.
- Set `APP_DOMAIN` to the public Foxel URL (for example, `https://my.foxel.cc`) to allow its Host and Origin headers through MCP transport validation. Loopback access remains available.
- `time`, directory listing, metadata, text reads, search, processor listing, and `web_fetch` can run directly. `web_fetch` supports GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS, including external write requests, without an additional Foxel approval step.
- External MCP clients can access file-system and processor tools. `time` and `web_fetch` are available only to the built-in AI agent and are not exposed through MCP.
- Directory listing, metadata, text reads, search, and processor listing can run directly. In the built-in agent, `time` and `web_fetch` also run without approval. `web_fetch` supports GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS, including external write requests, without an additional Foxel approval step.
- Agent writes, creates, deletes, moves, copies, renames, and `processors_run` calls create database-backed approval batches. Approval requests must include the returned `approval_batch_id`; execution uses the server-stored arguments, rejects expired or cross-user requests, and does not re-execute repeated confirmations.
- Recursive changes and directory processing validate all enumerated children before starting. Any denied or unverifiable path rejects the batch; queued work rechecks permissions and stops after a scope change.
- Tools return structured error codes such as `permission_denied`, `invalid_arguments`, `permission_scope_unverifiable`, `approval_expired`, `request_timeout`, and `execution_failed` without exposing internal exceptions or credentials.
+2 -1
View File
@@ -81,7 +81,8 @@ UC、123、115、天翼云盘、SMB、Seafile 和 Cloudreve V4 适配器无需
- MCP 服务地址为 `/api/mcp/`,外部客户端使用 Foxel Bearer Token;直接调用由客户端负责确认,服务端始终执行当前用户的路径权限。
- 将 `APP_DOMAIN` 设置为 Foxel 的公开访问地址(例如 `https://my.foxel.cc`),以允许对应的 Host 和 Origin 通过 MCP 传输校验。本机回环访问仍然可用。
- `time`、目录浏览、元信息、文本读取、搜索、处理器列表及 `web_fetch` 可直接调用。`web_fetch` 支持 GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS,包含外部写请求但不在 Foxel 内额外审批。
- 外部 MCP 客户端可访问文件系统与处理器工具。`time` 和 `web_fetch` 仅供内置 AI 助手使用,不通过 MCP 暴露。
- 目录浏览、元信息、文本读取、搜索和处理器列表可直接调用。内置 AI 助手中的 `time` 和 `web_fetch` 也无需审批;`web_fetch` 支持 GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS,包含外部写请求但不在 Foxel 内额外审批。
- Agent 对文件写入、创建、删除、移动、复制、重命名和 `processors_run` 创建数据库审批批次。审批请求必须携带返回的 `approval_batch_id`;参数以服务端保存的原始调用为准,过期或跨用户请求会拒绝,重复确认不会再次执行。
- 目录变更和目录处理在执行前检查全部可见子项,任一项缺少权限或无法完整枚举则整批拒绝。后台任务会重新检查权限,权限变化后停止后续操作。
- 工具错误使用结构化错误码,例如 `permission_denied`、`invalid_arguments`、`permission_scope_unverifiable`、`approval_expired`、`request_timeout` 和 `execution_failed`,不会回传底层异常或凭据。
+35 -31
View File
@@ -154,7 +154,17 @@ class FoxelMcpTokenVerifier:
MCP_SERVER = FoxelMCPServer(
name="Foxel MCP",
instructions="Foxel 内置 MCP 服务,提供文件系统、网页抓取、时间与处理器相关能力。",
instructions=(
"Foxel MCP provides private-cloud file browsing, search, text reading and writing, file organization, and processor tasks."
" All file paths are absolute paths in the Foxel virtual filesystem (e.g. /photos/a.jpg), not server-local paths."
" File operations enforce the authenticated account's path permissions."
" Use vfs_list_dir for known directories, vfs_search for unknown file locations, vfs_stat for metadata, and vfs_read_text for text."
" Read existing text and verify that it is complete before editing. Choose vfs_copy to preserve the source or vfs_move to relocate it."
" Before running a processor, use processors_list to discover available types, supported formats, and configuration."
" Tool results include ok; inspect error on failure. A task_id or queued=true indicates submission, not completion."
" Check progress in the Foxel task queue; this service does not provide a task-status tool."
" External MCP clients handle confirmation for file mutations. The built-in agent uses its own tool approval policy."
),
token_verifier=FoxelMcpTokenVerifier(),
auth=AuthSettings(
issuer_url="http://127.0.0.1:8000",
@@ -179,7 +189,7 @@ for descriptor in mcp_tool_descriptors():
"foxel://context/current-path",
name="current_path",
title="Current Path",
description="返回当前请求上下文里的文件管理目录。",
description="Read when the user refers to the current directory. Returns the Foxel path from x-foxel-current-path, or null if no valid path was supplied.",
mime_type="application/json",
)
def current_path_resource() -> dict[str, Any]:
@@ -190,7 +200,7 @@ def current_path_resource() -> dict[str, Any]:
"foxel://policy/tool-confirmation",
name="tool_confirmation_policy",
title="Tool Confirmation Policy",
description="返回 Foxel agent 对工具审批的策略。",
description="Read to understand confirmation policies for exposed tools. Lists read-only, unconfirmed, and confirmation-required tools, and distinguishes client confirmation from built-in agent approval.",
mime_type="application/json",
)
def tool_confirmation_policy_resource() -> dict[str, Any]:
@@ -198,8 +208,7 @@ def tool_confirmation_policy_resource() -> dict[str, Any]:
"read_tools": [tool.name for tool in mcp_tool_descriptors() if tool.annotations.get("readOnlyHint")],
"unconfirmed_tools": [tool.name for tool in mcp_tool_descriptors() if not tool.requires_confirmation],
"write_tools": [tool.name for tool in mcp_tool_descriptors() if tool.requires_confirmation],
"rule": "直接调用 MCP tool 时不额外审批;通过 agent 代表用户执行写操作时需要审批。",
"web_fetch": "所有 HTTP 方法免审批,包含可能修改外部系统的请求;免审批不代表只读。",
"rule": "Direct MCP calls do not require additional server approval; external clients handle confirmation. The built-in agent requires approval for confirmation-required tools unless auto-execution is enabled.",
}
@@ -207,7 +216,7 @@ def tool_confirmation_policy_resource() -> dict[str, Any]:
"foxel://processors/index",
name="processors_index",
title="Processors Index",
description="返回当前可用处理器列表。",
description="Read to choose a processor and check supported formats and configuration schemas. Returns the same available processor list as processors_list.",
mime_type="application/json",
)
def processors_index_resource() -> dict[str, Any]:
@@ -222,7 +231,7 @@ async def _tool_resource(tool_name: str, arguments: dict[str, Any], current_path
"foxel://vfs/stat/{path}",
name="vfs_stat_resource",
title="VFS Stat",
description="读取指定路径的文件或目录元信息;path 需要 URL 编码。",
description="Read to inspect file or directory type, size, and modification time. URL-encode the Foxel path without its leading /. Uses the same permission checks as vfs_stat.",
mime_type="application/json",
)
async def vfs_stat_resource(path: str, ctx: Context) -> dict[str, Any]:
@@ -233,7 +242,7 @@ async def vfs_stat_resource(path: str, ctx: Context) -> dict[str, Any]:
"foxel://vfs/text/{path}",
name="vfs_text_resource",
title="VFS Text",
description="读取文本文件内容;path 需要 URL 编码。",
description="Read a known text file. URL-encode the Foxel path without its leading /. Defaults to UTF-8 and up to 8000 characters; use vfs_read_text to adjust encoding or length.",
mime_type="application/json",
)
async def vfs_text_resource(path: str, ctx: Context) -> dict[str, Any]:
@@ -244,7 +253,7 @@ async def vfs_text_resource(path: str, ctx: Context) -> dict[str, Any]:
"foxel://vfs/dir/{path}",
name="vfs_dir_resource",
title="VFS Directory",
description="列出目录内容;path 需要 URL 编码。",
description="Read the first page of a known directory. URL-encode the Foxel path without its leading /. Use vfs_list_dir for pagination or sorting.",
mime_type="application/json",
)
async def vfs_dir_resource(path: str, ctx: Context) -> dict[str, Any]:
@@ -255,44 +264,39 @@ async def vfs_dir_resource(path: str, ctx: Context) -> dict[str, Any]:
"foxel://vfs/search/{query}",
name="vfs_search_resource",
title="VFS Search",
description="搜索文件;query 需要 URL 编码。",
description="Read to run default semantic search using a natural-language content description. URL-encode query. Requires a content vector index; use vfs_search for filename search or custom limits.",
mime_type="application/json",
)
async def vfs_search_resource(query: str, ctx: Context) -> dict[str, Any]:
return await _tool_resource("vfs_search", {"q": unquote(query)}, _header_current_path(ctx))
@MCP_SERVER.prompt(name="browse_path", title="Browse Path", description="生成浏览目录的推荐提示词。")
def browse_path_prompt(path: Annotated[str, Field(description="目标目录路径")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请先浏览目录 `{path}`,总结结构与关键文件。必要时调用 vfs_list_dir 与 vfs_stat。"}]
@MCP_SERVER.prompt(name="browse_path", title="Browse Path", description="Use to explore an unfamiliar directory and summarize its structure with vfs_list_dir and vfs_stat.")
def browse_path_prompt(path: Annotated[str, Field(description="Absolute Foxel directory path to explore.")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"Browse directory `{path}` and summarize its structure and key files. Use vfs_list_dir and vfs_stat as needed."}]
@MCP_SERVER.prompt(name="inspect_file", title="Inspect File", description="生成查看文件的推荐提示词。")
def inspect_file_prompt(path: Annotated[str, Field(description="目标文件路径")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请检查文件 `{path}` 的内容与用途。必要时调用 vfs_read_text。"}]
@MCP_SERVER.prompt(name="inspect_file", title="Inspect File", description="Use to read a known text file and explain its contents and purpose with vfs_read_text.")
def inspect_file_prompt(path: Annotated[str, Field(description="Absolute Foxel text file path to inspect.")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"Inspect the contents and purpose of file `{path}`. Use vfs_read_text as needed and check whether the content is truncated."}]
@MCP_SERVER.prompt(name="search_files", title="Search Files", description="生成搜索文件的推荐提示词。")
def search_files_prompt(query: Annotated[str, Field(description="搜索关键词")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请搜索与 `{query}` 相关的文件,并按相关性总结。必要时调用 vfs_search。"}]
@MCP_SERVER.prompt(name="search_files", title="Search Files", description="Use to find and summarize relevant files with vfs_search when their locations are unknown.")
def search_files_prompt(query: Annotated[str, Field(description="Content description or filename keywords to search for.")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"Search for files related to `{query}` and summarize them by relevance. Use vfs_search with the appropriate search mode."}]
@MCP_SERVER.prompt(name="edit_file_safely", title="Edit File Safely", description="生成安全修改文件的推荐提示词。")
def edit_file_safely_prompt(path: Annotated[str, Field(description="目标文件路径")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请先读取 `{path}`,解释拟修改点,再等待我确认后执行写入。"}]
@MCP_SERVER.prompt(name="edit_file_safely", title="Edit File Safely", description="Use to review an edit plan before replacing existing text: read first, explain changes, and wait for confirmation before writing.")
def edit_file_safely_prompt(path: Annotated[str, Field(description="Absolute Foxel text file path to edit.")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"Read `{path}` and verify that the content is complete. Explain the proposed edits, then wait for my confirmation before writing."}]
@MCP_SERVER.prompt(name="run_processor", title="Run Processor", description="生成运行处理器的推荐提示词。")
@MCP_SERVER.prompt(name="run_processor", title="Run Processor", description="Use with a known input path and processor type to check compatibility and confirm settings before submitting a processing task.")
def run_processor_prompt(
path: Annotated[str, Field(description="目标文件或目录路径")],
processor_type: Annotated[str, Field(description="处理器类型")],
path: Annotated[str, Field(description="Absolute Foxel file or directory path to process.")],
processor_type: Annotated[str, Field(description="Processor type from processors_list.")],
) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请检查 `{path}` 是否适合运行处理器 `{processor_type}`,确认参数后再执行 processors_run。"}]
@MCP_SERVER.prompt(name="fetch_web_page", title="Fetch Web Page", description="生成抓取网页的推荐提示词。")
def fetch_web_page_prompt(url: Annotated[str, Field(description="目标网址")]) -> list[dict[str, Any]]:
return [{"role": "user", "content": f"请抓取网页 `{url}`,并总结标题、正文与关键链接。必要时调用 web_fetch。"}]
return [{"role": "user", "content": f"Check processors_list and verify that `{path}` is suitable for processor `{processor_type}`. Confirm settings before calling processors_run. A task_id indicates submission, not completion."}]
_LOOPBACK_HOSTS = ["127.0.0.1", "localhost", "[::1]"]
+13 -3
View File
@@ -14,6 +14,8 @@ from models.database import AgentApprovalBatch, AgentApprovalCall
from .approvals import create_batch, execute_call, load_batch, pending_call, result_message
from .mcp import mcp_client_session, mcp_content_to_text
from .execution import execute_tool
from .tools import AGENT_ONLY_TOOL_NAMES, mcp_tool_descriptors
from .tools.base import tool_error, tool_result_to_content
from .types import AgentChatRequest
@@ -75,9 +77,15 @@ def _sse(event: str, data: Any) -> bytes:
async def _list_mcp_tools(session) -> list[dict[str, Any]]:
result = await session.list_tools()
return [{"name": item.name, "description": item.description or "", "input_schema": item.input_schema or {},
"annotations": item.annotations.model_dump(exclude_none=True) if item.annotations else {},
"meta": item.meta or {}} for item in result.tools]
tools = [{"name": item.name, "description": item.description or "", "input_schema": item.input_schema or {},
"annotations": item.annotations.model_dump(exclude_none=True) if item.annotations else {},
"meta": item.meta or {}} for item in result.tools]
tools.extend(
{"name": item.name, "description": item.description, "input_schema": item.input_schema,
"annotations": item.annotations, "meta": item.meta}
for item in mcp_tool_descriptors(include_agent_only=True) if item.name in AGENT_ONLY_TOOL_NAMES
)
return tools
async def _execute_mcp_call(session, name: str, arguments: dict[str, Any]) -> str:
@@ -141,6 +149,8 @@ class AgentService:
index = {tool["name"]: tool for tool in tools}
async def execute(name, arguments):
if name in AGENT_ONLY_TOOL_NAMES:
return tool_result_to_content(await execute_tool(name, arguments, user, path))
return await _execute_mcp_call(session, name, arguments)
if validated:
+8 -2
View File
@@ -10,6 +10,8 @@ TOOLS: Dict[str, ToolSpec] = {}
for group in (TIME_TOOLS, WEB_FETCH_TOOLS, PROCESSOR_TOOLS, VFS_TOOLS):
TOOLS.update(group)
AGENT_ONLY_TOOL_NAMES = frozenset({"time", "web_fetch"})
def get_tool(name: str) -> Optional[ToolSpec]:
return TOOLS.get(name)
@@ -19,11 +21,15 @@ def list_tool_specs() -> List[ToolSpec]:
return list(TOOLS.values())
def mcp_tool_descriptors() -> List[McpToolDescriptor]:
return [tool_spec_to_mcp_descriptor(spec) for spec in TOOLS.values()]
def mcp_tool_descriptors(*, include_agent_only: bool = False) -> List[McpToolDescriptor]:
return [
tool_spec_to_mcp_descriptor(spec) for spec in TOOLS.values()
if include_agent_only or spec.name not in AGENT_ONLY_TOOL_NAMES
]
__all__ = [
"AGENT_ONLY_TOOL_NAMES",
"McpToolDescriptor",
"ToolSpec",
"get_tool",
+19 -11
View File
@@ -64,7 +64,11 @@ async def _processors_run(args: Dict[str, Any]) -> Dict[str, Any]:
TOOLS: Dict[str, ToolSpec] = {
"processors_list": ToolSpec(
name="processors_list",
description="获取可用处理器列表(type/name/config_schema 等)。",
description=(
"Use before watermarking images, converting files, or performing other processing tasks to discover available processors."
" Returns type, name, supported_exts, config_schema, produces_file, and supports_directory."
" Choose a listed processor_type compatible with the input, build config from config_schema, then call processors_run."
),
parameters={
"type": "object",
"properties": {},
@@ -76,20 +80,24 @@ TOOLS: Dict[str, ToolSpec] = {
"processors_run": ToolSpec(
name="processors_run",
description=(
"运行处理器处理文件或目录。"
" 对目录可选 max_depth/suffix;对文件可选 overwrite/save_to。"
" 返回任务 id(去任务队列查看进度)。"
"Submit a processing task for a Foxel file or directory, such as watermarking, conversion, or indexing;"
" available capabilities are listed by processors_list. Check processor settings and formats first, and verify the input with vfs_stat."
" For processors that produce files, preserve a single input with overwrite=false and save_to, or replace it with overwrite=true."
" Set max_depth or suffix to enable per-file directory batching; max_depth=0 processes only the current level."
" Batching defaults to overwrite=true; when producing files with overwrite=false, suffix is required to save copies."
" Directories without max_depth/suffix use the processor's regular entry point, subject to supports_directory and overwrite."
" A returned task_id means the task was submitted, not completed. Check results in the Foxel task queue."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件或目录路径(绝对路径,如 /foo/bar)"},
"processor_type": {"type": "string", "description": "处理器类型(例如 image_watermark)"},
"config": {"type": "object", "description": "处理器配置,按 processors_list 返回的 config_schema 填写"},
"overwrite": {"type": "boolean", "description": "是否覆盖原文件/目录内文件"},
"save_to": {"type": "string", "description": "保存到指定路径(仅文件模式,且 overwrite=false 时使用)"},
"max_depth": {"type": "integer", "minimum": 0, "description": "目录遍历深度(仅目录模式)"},
"suffix": {"type": "string", "description": "目录批处理时的输出后缀(仅 produces_file 且 overwrite=false)"},
"path": {"type": "string", "description": "Absolute Foxel virtual file or directory path, e.g. /photos/image.jpg or /photos."},
"processor_type": {"type": "string", "description": "A type returned by processors_list, e.g. image_watermark. Check supported formats and directory capabilities first."},
"config": {"type": "object", "description": "Processor configuration following the config_schema returned by processors_list."},
"overwrite": {"type": "boolean", "description": "Whether to replace input files. Defaults to false for single files and regular directory processing, true for directory batching with max_depth or suffix. Specify explicitly."},
"save_to": {"type": "string", "description": "Full absolute Foxel output path for a single file with overwrite=false. Required when producing a file while preserving the input; invalid for directories."},
"max_depth": {"type": "integer", "minimum": 0, "description": "Subdirectory depth for per-file batching: 0 for the current level, 1 to include one nested level. Omit for unlimited depth. Providing this enables directory batching."},
"suffix": {"type": "string", "description": "Output filename suffix for directory batching, e.g. _watermarked creates photo_watermarked.jpg. Required when produces_file=true and overwrite=false. Must not contain path separators."},
},
"required": ["path", "processor_type"],
},
+12 -8
View File
@@ -71,18 +71,22 @@ TOOLS: Dict[str, ToolSpec] = {
"time": ToolSpec(
name="time",
description=(
"获取服务器当前时间(精确到秒,含英文星期)。"
" 支持 year/month/day/hour/minute/second 偏移(可为负数)。"
"Use to get the current date, time, or weekday, or resolve relative dates such as tomorrow or last month."
" Returns server-local time to the second and the weekday in English; omit all arguments for the current time."
" year/month/day/hour/minute/second are signed offsets from now and can be combined;"
" day=1 means this time tomorrow and month=-1 means this time last month."
" Year and month offsets apply first, clamping the day to the target month's last day,"
" followed by day/hour/minute/second offsets. Does not convert time zones."
),
parameters={
"type": "object",
"properties": {
"year": {"type": "integer", "description": "年偏移(可为负数)"},
"month": {"type": "integer", "description": "月偏移(可为负数)"},
"day": {"type": "integer", "description": "日偏移(可为负数)"},
"hour": {"type": "integer", "description": "时偏移(可为负数)"},
"minute": {"type": "integer", "description": "分偏移(可为负数)"},
"second": {"type": "integer", "description": "秒偏移(可为负数)"},
"year": {"type": "integer", "description": "Year offset from now; may be negative. Default: 0."},
"month": {"type": "integer", "description": "Month offset from now; may be negative. Default: 0."},
"day": {"type": "integer", "description": "Day offset from now; may be negative. Default: 0."},
"hour": {"type": "integer", "description": "Hour offset from now; may be negative. Default: 0."},
"minute": {"type": "integer", "description": "Minute offset from now; may be negative. Default: 0."},
"second": {"type": "integer", "description": "Second offset from now; may be negative. Default: 0."},
},
"additionalProperties": False,
},
+84 -39
View File
@@ -148,16 +148,20 @@ async def _vfs_search(args: Dict[str, Any]) -> Dict[str, Any]:
TOOLS: Dict[str, ToolSpec] = {
"vfs_list_dir": ToolSpec(
name="vfs_list_dir",
description="浏览目录(列出 entries + pagination)。",
description=(
"Use to browse a known directory, understand its structure, or find files within it. Returns entries and pagination."
" Start at / to discover accessible roots. Lists one level only; call again to explore subdirectories."
" Follow pagination to retrieve more entries. Use vfs_search first when the file location is unknown."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "目录路径(绝对路径,如 /foo/bar)"},
"page": {"type": "integer", "minimum": 1, "description": "页码(从 1 开始)"},
"page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "每页条数"},
"sort_by": {"type": "string", "enum": ["name", "size", "mtime"], "description": "排序字段:name/size/mtime"},
"sort_order": {"type": "string", "enum": ["asc", "desc"], "description": "排序顺序:asc/desc"},
"cursor": {"type": "string", "description": "游标分页的下一页游标"},
"path": {"type": "string", "description": "Absolute Foxel virtual directory path, e.g. / or /photos; not a server-local path."},
"page": {"type": "integer", "minimum": 1, "description": "Page number starting at 1. Default: 1. Increment for page-based pagination."},
"page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "Entries per page. Default: 50. Maximum: 100."},
"sort_by": {"type": "string", "enum": ["name", "size", "mtime"], "description": "Sort by name, size, or modification time (mtime). Default: name."},
"sort_order": {"type": "string", "enum": ["asc", "desc"], "description": "Sort direction. Default: asc. Use mtime with desc for recently modified files."},
"cursor": {"type": "string", "description": "For cursor pagination, pass the previous response's next_cursor unchanged. Keep path, sorting, and page_size the same."},
},
"required": ["path"],
"additionalProperties": False,
@@ -167,11 +171,15 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_stat": ToolSpec(
name="vfs_stat",
description="查看文件/目录信息(size/mtime/is_dir/has_thumbnail/vector_index 等)。",
description=(
"Use to check whether a path exists, distinguish files from directories, or inspect size and modification time."
" Returns metadata such as size, mtime, and is_dir; extra fields depend on the storage adapter."
" Verify a target before reading, writing, or running a processor. Use vfs_read_text for text content."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "路径(绝对路径,如 /foo/bar.txt)"},
"path": {"type": "string", "description": "Absolute Foxel virtual file or directory path, e.g. /docs/report.txt."},
},
"required": ["path"],
"additionalProperties": False,
@@ -181,13 +189,18 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_read_text": ToolSpec(
name="vfs_read_text",
description="读取文本文件内容(解码失败视为二进制,返回 error)。",
description=(
"Use to read, summarize, or inspect text files such as Markdown, configuration, or source code before editing."
" Returns content, truncated, and the original character count (length). Defaults to the first 8000 characters."
" When truncated=true, content is incomplete; increase max_chars and reread, up to 100000. Does not support paged reads."
" Decoding failures return an error; specify encoding when known. Not suitable for images or other binary files."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径(绝对路径,如 /foo/bar.md)"},
"encoding": {"type": "string", "description": "文本编码(默认 utf-8)"},
"max_chars": {"type": "integer", "minimum": 1, "maximum": 100000, "description": "最多返回的字符数(默认 8000)"},
"path": {"type": "string", "description": "Absolute Foxel virtual text file path, e.g. /docs/report.md."},
"encoding": {"type": "string", "description": "Text encoding. Default: utf-8."},
"max_chars": {"type": "integer", "minimum": 1, "maximum": 100000, "description": "Maximum characters returned from the start of the file. Default: 8000. Limit: 100000. Increase and reread if truncated."},
},
"required": ["path"],
"additionalProperties": False,
@@ -197,13 +210,17 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_write_text": ToolSpec(
name="vfs_write_text",
description="写入文本文件内容(会覆盖目标文件)。",
description=(
"Use to create a text file or save its complete edited contents. Replaces an existing file in full."
" Before editing an existing file, read it with vfs_read_text and check truncated to avoid overwriting it with incomplete content."
" content must contain the complete final text; an empty string clears the file. Does not append or perform partial replacements."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径(绝对路径,如 /foo/bar.md)"},
"content": {"type": "string", "description": "要写入的文本内容"},
"encoding": {"type": "string", "description": "文本编码(默认 utf-8)"},
"path": {"type": "string", "description": "Absolute Foxel virtual text file path, e.g. /docs/report.md."},
"content": {"type": "string", "description": "Complete final text replacing any existing contents. An empty string clears the file."},
"encoding": {"type": "string", "description": "Text encoding. Default: utf-8."},
},
"required": ["path", "content"],
"additionalProperties": False,
@@ -213,11 +230,15 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_mkdir": ToolSpec(
name="vfs_mkdir",
description="创建目录。",
description=(
"Use to create a Foxel virtual directory for organizing files or preparing an output location."
" path is the complete new directory path. Handling of missing parents and existing directories depends on the storage adapter."
" Check the target and parent with vfs_stat or vfs_list_dir first if needed."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "目录路径(绝对路径,如 /foo/bar)"},
"path": {"type": "string", "description": "Full absolute Foxel virtual directory path to create, e.g. /docs/archive."},
},
"required": ["path"],
"additionalProperties": False,
@@ -227,11 +248,15 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_delete": ToolSpec(
name="vfs_delete",
description="删除文件或目录(由底层适配器决定是否递归)。",
description=(
"Use when the user requests removal of a file or directory. Directory deletion may remove contents recursively, depending on the storage adapter."
" Verify the full path before calling; inspect the deletion scope with vfs_stat and vfs_list_dir if needed."
" Does not guarantee a recycle bin or recovery. Cannot delete the virtual root /."
),
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "路径(绝对路径,如 /foo/bar 或 /foo/bar.txt)"},
"path": {"type": "string", "description": "Absolute Foxel virtual path to delete, e.g. /docs/archive or /docs/old.txt."},
},
"required": ["path"],
"additionalProperties": False,
@@ -241,13 +266,18 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_move": ToolSpec(
name="vfs_move",
description="移动路径(可能进入任务队列)。",
description=(
"Use to relocate a file or directory, reorganize folders, or transfer between storage mounts."
" src and dst are full paths; dst must include the final file or directory name. The source is removed after successful completion."
" Does not overwrite by default. overwrite=true may replace the destination and its directory contents."
" queued=true with task_id means the operation was queued, not completed. Confirm completion in the Foxel task queue before reporting success."
),
parameters={
"type": "object",
"properties": {
"src": {"type": "string", "description": "源路径(绝对路径)"},
"dst": {"type": "string", "description": "目标路径(绝对路径)"},
"overwrite": {"type": "boolean", "description": "是否允许覆盖已存在目标(默认 false)"},
"src": {"type": "string", "description": "Absolute Foxel virtual source path."},
"dst": {"type": "string", "description": "Full absolute destination path including the final file or directory name, e.g. /archive/report.md."},
"overwrite": {"type": "boolean", "description": "Default: false. true allows replacing an existing destination, potentially including all contents of a destination directory."},
},
"required": ["src", "dst"],
"additionalProperties": False,
@@ -257,13 +287,18 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_copy": ToolSpec(
name="vfs_copy",
description="复制路径(可能进入任务队列)。",
description=(
"Use to create a copy or backup, including across storage mounts, while preserving the source file or directory."
" dst is the full destination path including the final file or directory name."
" Does not overwrite by default. overwrite=true may replace the destination and its directory contents."
" queued=true with task_id means the operation was queued, not completed. Confirm completion in the Foxel task queue before reporting success."
),
parameters={
"type": "object",
"properties": {
"src": {"type": "string", "description": "源路径(绝对路径)"},
"dst": {"type": "string", "description": "目标路径(绝对路径)"},
"overwrite": {"type": "boolean", "description": "是否覆盖已存在目标(默认 false)"},
"src": {"type": "string", "description": "Absolute Foxel virtual source path."},
"dst": {"type": "string", "description": "Full absolute copy destination including the final file or directory name, e.g. /backup/report.md."},
"overwrite": {"type": "boolean", "description": "Default: false. true allows replacing an existing destination, potentially including all contents of a destination directory."},
},
"required": ["src", "dst"],
"additionalProperties": False,
@@ -273,13 +308,17 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_rename": ToolSpec(
name="vfs_rename",
description="重命名路径(本质是同目录 move)。",
description=(
"Use to change a file or directory name; keep src and dst in the same parent directory when renaming."
" dst is the full path including the new name. Does not overwrite an existing destination by default."
" Only supports renaming within one storage mount. Use vfs_move for relocation or transfers across mounts."
),
parameters={
"type": "object",
"properties": {
"src": {"type": "string", "description": "源路径(绝对路径)"},
"dst": {"type": "string", "description": "目标路径(绝对路径)"},
"overwrite": {"type": "boolean", "description": "是否允许覆盖已存在目标(默认 false)"},
"src": {"type": "string", "description": "Absolute Foxel virtual source path."},
"dst": {"type": "string", "description": "Full absolute path including the new name, e.g. rename /docs/old.md to /docs/new.md."},
"overwrite": {"type": "boolean", "description": "Default: false. true allows replacing an existing destination, potentially including all contents of a destination directory."},
},
"required": ["src", "dst"],
"additionalProperties": False,
@@ -289,15 +328,21 @@ TOOLS: Dict[str, ToolSpec] = {
),
"vfs_search": ToolSpec(
name="vfs_search",
description="搜索文件(mode=vector 或 filename)。",
description=(
"Use to find files by topic or name when their location is unknown. Returns only matches the current user can read."
" Default vector mode supports natural-language semantic queries and requires an embedding model and content vector index."
" filename mode matches filename or path keywords and supports page/page_size pagination."
" Both modes depend on existing indexes; empty results do not prove a file is absent. Browse known directories with vfs_list_dir."
" Verify returned paths with vfs_stat or vfs_read_text. Search snippets are not complete file contents."
),
parameters={
"type": "object",
"properties": {
"q": {"type": "string", "description": "搜索关键词"},
"mode": {"type": "string", "enum": ["vector", "filename"], "description": "搜索模式:vector/filename(默认 vector)"},
"top_k": {"type": "integer", "minimum": 1, "maximum": 100, "description": "返回数量(vector 模式使用,默认 10)"},
"page": {"type": "integer", "minimum": 1, "description": "页码(filename 模式使用,默认 1)"},
"page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "分页大小(filename 模式使用,默认 10)"},
"q": {"type": "string", "description": "Nonempty query. For vector mode, describe content naturally, e.g. beach photos from last year's trip. For filename mode, use filename or path keywords, e.g. report.pdf."},
"mode": {"type": "string", "enum": ["vector", "filename"], "description": "vector for semantic content search (default), filename for filename or path keyword search. Both require existing indexes."},
"top_k": {"type": "integer", "minimum": 1, "maximum": 100, "description": "Maximum results in vector mode. Default: 10. Limit: 100."},
"page": {"type": "integer", "minimum": 1, "description": "Page number in filename mode. Default: 1."},
"page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "Results per page in filename mode. Default: 10. Limit: 100."},
},
"required": ["q"],
"additionalProperties": False,
+12 -8
View File
@@ -170,18 +170,22 @@ TOOLS: Dict[str, ToolSpec] = {
"web_fetch": ToolSpec(
name="web_fetch",
description=(
"抓取网页内容,返回状态、标题、正文、HTML、链接等信息。"
" 支持 GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS。"
"Use to read a known webpage URL, extract text and links, or call an HTTP API. Defaults to GET and follows redirects."
" HTML responses include title/text/html/links; other responses include decoded text."
" Check status_code to determine HTTP success; content_type and final_url are also returned."
" Does not execute JavaScript, render browser-dependent content, or search the web."
" Supports GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS without tool approval; write methods may change external systems."
" HTTP timeout: 20 seconds. Response body limit: 2 MiB; larger responses return an error."
),
parameters={
"type": "object",
"properties": {
"url": {"type": "string", "description": "目标 URL"},
"method": {"type": "string", "description": "请求方法(默认 GET,所有方法免审批)"},
"headers": {"type": "object", "description": "请求头", "additionalProperties": {"type": "string"}},
"params": {"type": "object", "description": "查询参数", "additionalProperties": {"type": "string"}},
"json": {"type": "object", "description": "JSON 请求体"},
"body": {"type": "string", "description": "原始请求体"},
"url": {"type": "string", "description": "Full http:// or https:// URL, e.g. https://example.com/article. Must not be search keywords."},
"method": {"type": "string", "description": "HTTP method: GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS. Default: GET. All methods bypass tool approval."},
"headers": {"type": "object", "description": "Request headers with string values, e.g. Accept or Content-Type, as required by the target API.", "additionalProperties": {"type": "string"}},
"params": {"type": "object", "description": "URL query parameters with string values, e.g. {\"page\": \"1\"}.", "additionalProperties": {"type": "string"}},
"json": {"type": "object", "description": "JSON object request body. Takes precedence when body is also provided."},
"body": {"type": "string", "description": "Raw text request body, used only when json is omitted. Set Content-Type in headers if needed."},
},
"required": ["url"],
"additionalProperties": False,
+44 -4
View File
@@ -17,7 +17,7 @@ from domain.agent.api import router
from domain.agent.approvals import create_batch, execute_call, load_batch, now
from domain.agent.execution import execute_tool, snapshot_tree, validate_arguments
from domain.agent.mcp import MCP_HTTP_APP, mcp_client_session
from domain.agent.service import AgentService, _choose_chat_ability
from domain.agent.service import AgentService, _choose_chat_ability, _list_mcp_tools
from domain.agent.tools import mcp_tool_descriptors
from domain.agent.tools.base import tool_error, tool_result_to_content
from domain.agent.types import AgentChatRequest
@@ -261,7 +261,7 @@ class BoundaryTests(DatabaseTests):
processed.assert_not_awaited()
async def test_web_methods_remain_unconfirmed_with_accurate_annotations(self):
descriptor = next(d for d in mcp_tool_descriptors() if d.name == "web_fetch")
descriptor = next(d for d in mcp_tool_descriptors(include_agent_only=True) if d.name == "web_fetch")
self.assertFalse(descriptor.requires_confirmation)
self.assertFalse(descriptor.annotations["readOnlyHint"])
self.assertTrue(descriptor.annotations["destructiveHint"])
@@ -292,7 +292,23 @@ class BoundaryTests(DatabaseTests):
async with MCP_HTTP_APP.router.lifespan_context(MCP_HTTP_APP):
async with mcp_client_session(self.user, "//allowed//docs") as session:
tools = await session.list_tools()
self.assertEqual(len(tools.tools), 14)
self.assertEqual(len(tools.tools), 12)
self.assertFalse({"time", "web_fetch"} & {tool.name for tool in tools.tools})
agent_tools = await _list_mcp_tools(session)
self.assertEqual(len(agent_tools), 14)
self.assertTrue({"time", "web_fetch"} <= {tool["name"] for tool in agent_tools})
for name, arguments in (("time", {}), ("web_fetch", {"url": "https://example.com"})):
with self.subTest(tool=name), patch("domain.agent.mcp.execute_tool", AsyncMock()) as executed:
blocked = await session.call_tool(name, arguments)
self.assertTrue(blocked.is_error)
executed.assert_not_awaited()
prompts = await session.list_prompts()
self.assertNotIn("fetch_web_page", {prompt.name for prompt in prompts.prompts})
policy = await session.read_resource("foxel://policy/tool-confirmation")
policy_data = json.loads(policy.contents[0].text)
self.assertNotIn("web_fetch", policy_data)
for key in ("read_tools", "unconfirmed_tools", "write_tools"):
self.assertFalse({"time", "web_fetch"} & set(policy_data[key]))
read_tool = next(tool for tool in tools.tools if tool.name == "vfs_read_text")
bounds = read_tool.input_schema["properties"]["max_chars"]
self.assertIn("100000", json.dumps(bounds))
@@ -313,7 +329,7 @@ class ApprovalTests(DatabaseTests):
{"id": "one", "name": "vfs_write_text", "arguments": {"path": "/allowed/one", "content": "original"}},
{"id": "two", "name": "vfs_mkdir", "arguments": {"path": "/allowed/two"}},
]}
self.index = {d.name: {"meta": d.meta} for d in mcp_tool_descriptors()}
self.index = {d.name: {"meta": d.meta} for d in mcp_tool_descriptors(include_agent_only=True)}
self.executed = []
async def call_tool(name, arguments):
@@ -474,6 +490,30 @@ class ApprovalTests(DatabaseTests):
self.assertEqual(self.model.await_count, 8)
self.assertEqual(len([m for m in result["messages"] if m["role"] == "tool"]), 8)
async def test_agent_only_tools_execute_locally_without_approval(self):
self.model.side_effect = [
{"role": "assistant", "content": "", "mcp_calls": [
{"id": "clock", "name": "time", "arguments": {}},
{"id": "page", "name": "web_fetch", "arguments": {"url": "https://example.com"}},
{"id": "invalid", "name": "web_fetch", "arguments": {"url": "file:///etc/passwd"}},
]},
{"role": "assistant", "content": "done"},
]
original_client = httpx.AsyncClient
transport = httpx.MockTransport(lambda req: httpx.Response(200, text="test page", request=req))
with patch("domain.agent.tools.web_fetch.httpx.AsyncClient", lambda **kw: original_client(transport=transport, **kw)):
result = await AgentService.chat(AgentChatRequest(messages=[{"role": "user", "content": "check time and page"}]), self.user)
self.assertEqual(result["finish_reason"], "completed")
self.assertEqual(result["pending_mcp_calls"], [])
self.assertEqual(self.executed, [])
outputs = {message["mcp_call_id"]: json.loads(message["content"])
for message in result["messages"] if message["role"] == "tool"}
self.assertTrue(outputs["clock"]["ok"])
self.assertIn("datetime", outputs["clock"]["data"])
self.assertTrue(outputs["page"]["ok"])
self.assertEqual(outputs["page"]["data"]["text"], "test page")
self.assert_error(outputs["invalid"], "invalid_arguments")
async def test_missing_model_and_unsupported_provider_are_explicit(self):
self.patches[-1].stop()
with patch("domain.agent.service.AIProviderService.get_default_model", AsyncMock(return_value=None)):