MCP 实战 08：查询超时怎么办？给工具加上有限重试

MCP 实战 09：同一个工具，怎么让不同的人只能查自己的项目？

工具能查到数据了，接下来还要问一句：

谁在查？他能看哪些数据？

比如 Alice 负责网页项目，Bob 负责移动项目。不能因为两个人用了同一个工具，就把所有记录都返回。

这篇用本地配置演示权限过滤，不需要账号系统。

一、先把规则说清楚

我们给两个人分配不同范围：

工具只接受项目参数。

身份由启动配置提供，不能让模型在调用工具时自己填。

否则，模型传一句“我是管理员”，程序就当真了，权限也就失去意义。

二、准备项目

新建文件夹 mcp-permissions：

mcp-permissions/
├── run.py
├── check.py
├── make-config.py
└── fixtures/
└── issues.json

在这个文件夹打开终端。

Mac 安装依赖：

python3 -m venv .venv
.venv/bin/python -m pip install mcp==2.3.0 pydantic==2.14.1

Windows PowerShell：

py –3 –m venv .venv
.\\.venv\\Scripts\\python.exe –m pip install mcp==2.3.0 pydantic==2.14.1

Python 使用 3.11 或更高版本。

三、放入三条演示数据

打开 fixtures/issues.json，复制：

[
{
"id": "WEB-1",
"owner": "alice",
"project": "web",
"title": "演示网页问题"
},
{
"id": "MOB-1",
"owner": "bob",
"project": "mobile",
"title": "演示移动问题"
},
{
"id": "WEB-PRIVATE",
"owner": "bob",
"project": "web",
"title": "另一个人的记录"
}
]

第三条是故意放进去的。

它也属于 web 项目，但归 Bob 所有。Alice 查询网页项目时，不应该看到这条记录。

四、在工具里检查身份和范围

打开 run.py，复制：

import json
import os
from pathlib import Path
from typing import Any, Literal

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

BASE = Path(__file__).resolve().parent

# 身份来自本地启动配置，不来自工具参数。
OWNER = os.environ.get("DEMO_OWNER", "")

SCOPES = {
"alice": {"web"},
"bob": {"mobile"},
}

server = MCPServer("local-scope-demo")

def query(project):
if OWNER not in SCOPES:
raise ToolError("untrusted or missing launcher identity")

if project not in SCOPES[OWNER]:
raise ToolError("project outside configured scope")

rows = json.loads(
(BASE / "fixtures/issues.json").read_text("utf-8")
)

return {
"owner": OWNER,
"project": project,
"rows": [
row for row in rows
if row["owner"] == OWNER
and row["project"] == project
],
}

@server.tool()
def query_issues(
project: Literal["web", "mobile"],
) –> dict[str, Any]:
"""只查询当前进程身份有权访问、且归其所有的演示记录。"""
return query(project)

if __name__ == "__main__":
server.run()

查询分三步：

这里没有 owner 工具参数，模型只能选择查询哪个项目。

五、检查“能查”和“不能查”

打开 check.py，复制完整代码：

import asyncio
import json
import os
from pathlib import Path
import sys

from mcp import Client, StdioServerParameters

BASE = Path(__file__).resolve().parent
checks = []

def check(name, condition):
assert condition, name
checks.append(name)

def config(owner):
return StdioServerParameters(
command=sys.executable,
args=[str(BASE / "run.py")],
cwd=str(BASE),
env={
**os.environ,
"PYTHONUTF8": "1",
"DEMO_OWNER": owner,
},
)

async def main():
results = {}

for owner, allowed, denied, expected in [
("alice", "web", "mobile", "WEB-1"),
("bob", "mobile", "web", "MOB-1"),
]:
async with Client(config(owner)) as client:
tool = (await client.list_tools()).tools[0]

check(
owner + "_schema_has_no_identity",
set(tool.input_schema["properties"]) == {"project"},
)

result = await client.call_tool(
"query_issues", {"project": allowed}
)
data = result.structured_content

check(
owner + "_allowed",
not result.is_error
and [row["id"] for row in data["rows"]] == [expected],
)
results[owner] = data

result = await client.call_tool(
"query_issues", {"project": denied}
)
check(owner + "_denied", result.is_error)

for owner in ["", "mallory"]:
async with Client(config(owner)) as client:
result = await client.call_tool(
"query_issues", {"project": "web"}
)
check(
"missing_identity" if not owner else "unknown_identity",
result.is_error,
)

async with Client(config("alice")) as a, Client(config("bob")) as b:
ra, rb = await asyncio.gather(
a.call_tool("query_issues", {"project": "web"}),
b.call_tool("query_issues", {"project": "mobile"}),
)
check(
"separate_process_isolation",
ra.structured_content["owner"] == "alice"
and rb.structured_content["owner"] == "bob",
)

report = {
"checks_passed": len(checks),
"checks": checks,
"all_passed": True,
"transport": "real stdio, separately configured processes",
"source": "fictional local fixture",
}

(BASE / "verification.json").write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\\n",
"utf-8",
)
(BASE / "demo-result.json").write_text(
json.dumps({
"checks_passed": len(checks),
"results": results,
"denied": "alice→mobile / bob→web",
"note": "本地可信配置身份演示",
}, ensure_ascii=False, indent=2) + "\\n",
"utf-8",
)

print(json.dumps(report, ensure_ascii=False))

if __name__ == "__main__":
asyncio.run(main())

Mac 执行：

.venv/bin/python check.py

Windows 执行：

.\\.venv\\Scripts\\python.exe check.py

正常结果包含：

{
"checks_passed": 9,
"all_passed": true
}

检查过程中会故意触发四次拒绝，所以出现权限错误是正常的。

打开运行后生成的 demo-result.json，可以看到：

• Alice 只拿到 WEB-1。
• Bob 只拿到 MOB-1。
• WEB-PRIVATE 没有混进 Alice 的结果。

六、生成两份客户端配置

打开 make-config.py，复制：

import json
from pathlib import Path
import sys

BASE = Path(__file__).resolve().parent

for owner in ["alice", "bob"]:
config = {
"mcpServers": {
"local-" + owner: {
"command": sys.executable,
"args": [str(BASE / "run.py")],
"env": {
"PYTHONUTF8": "1",
"DEMO_OWNER": owner,
},
}
}
}

target = BASE / ("client-" + owner + ".json")
target.write_text(
json.dumps(config, ensure_ascii=False, indent=2) + "\\n",
"utf-8",
)
print(target.name)

Mac 执行：

.venv/bin/python make-config.py

Windows 执行：

.\\.venv\\Scripts\\python.exe make-config.py

会生成两份配置：

client-alice.json
client-bob.json

按照第 04 篇的方法，从对应配置中取出 command、args、env，手动填写客户端。

先只给当前 Agent 启用 Alice 的服务，发送：

请调用 query_issues，project 填 web。
列出工具返回的问题编号和标题。

应该只看到 WEB-1。

再让它查询 mobile，应该被拒绝。不要同时启用 Bob 的服务，否则模型可能改用 Bob 的工具完成查询，影响观察结果。

七、这套演示的边界在哪？

这里的“可信身份”，意思是：启动配置由你或管理员控制，模型只能调用工具，不能修改配置。

它没有实现登录、OAuth 或远程用户认证。每个服务进程只有一个固定身份，不能拿一个全局变量同时代表所有用户。

如果使用者能够任意修改环境变量、代码或数据文件，就能改变演示规则。因此，这篇适合学习本地权限过滤，不能直接当成多人共享服务的安全隔离方案。