1

pip install fastapi uvicorn casbin pydantic python-multipart

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import os
import casbin
from typing import Optional, Dict, Any

app = FastAPI(title="MCP本地文件系统服务")

# 初始化Casbin权限引擎
# rbac_model.conf是权限模型定义，rbac_policy.csv是权限规则配置
e = casbin.Enforcer("rbac_model.conf", "rbac_policy.csv")

# 工作目录隔离，所有操作都限制在这个目录下
WORKSPACE_ROOT = "/home/ubuntu/agent/workspace"
os.makedirs(WORKSPACE_ROOT, exist_ok=True)

def sanitize_path(path: str) -> str:
    """路径清洗，防止目录遍历攻击"""
    abs_path = os.path.abspath(os.path.join(WORKSPACE_ROOT, path.lstrip("/")))
    if not abs_path.startswith(WORKSPACE_ROOT):
        raise HTTPException(status_code=403, detail="禁止访问工作目录外的路径")
    return abs_path

# MCP能力发现接口：返回当前服务支持的所有操作
@app.get("/mcp/capabilities")
def get_capabilities():
    return {
        "version": "1.0",
        "service_name": "local_file_system",
        "capabilities": [
            {
                "name": "file_list",
                "description": "列出指定目录下的所有文件和文件夹",
                "parameters": {
                    "path": {"type": "string", "description": "要列出的目录路径，相对于工作根目录"}
                },
                "returns": {"type": "array", "items": {"type": "string"}, "description": "文件和文件夹名称列表"}
            },
            {
                "name": "file_read",
                "description": "读取指定文本文件的内容",
                "parameters": {
                    "path": {"type": "string", "description": "要读取的文件路径，相对于工作根目录"},
                    "max_size": {"type": "integer", "default": 1048576, "description": "最大读取字节数，默认1MB，超过则返回截断内容"}
                },
                "returns": {"type": "string", "description": "文件内容"}
            },
            {
                "name": "file_write",
                "description": "写入内容到指定文件",
                "parameters": {
                    "path": {"type": "string", "description": "要写入的文件路径，相对于工作根目录"},
                    "content": {"type": "string", "description": "要写入的文本内容"},
                    "overwrite": {"type": "boolean", "default": False, "description": "如果文件已存在，是否覆盖"}
                },
                "returns": {"type": "object", "properties": {"status": {"type": "string"}, "size": {"type": "integer"}}, "description": "写入结果"}
            },
            {
                "name": "file_delete",
                "description": "删除指定文件",
                "parameters": {
                    "path": {"type": "string", "description": "要删除的文件路径，相对于工作根目录"}
                },
                "returns": {"type": "object", "properties": {"status": {"type": "string"}}, "description": "删除结果"}
            }
        ]
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

class MCPExecuteRequest(BaseModel):
    capability: str
    parameters: Dict[str, Any]
    context: Optional[Dict[str, Any]] = None

@app.post("/mcp/execute")
def execute_operation(request: MCPExecuteRequest, user_id: str = "default_agent"):
    # 路径清洗
    raw_path = request.parameters.get("path", "")
    safe_path = sanitize_path(raw_path)
    
    # 权限校验
    if not e.enforce(user_id, raw_path, request.capability):
        raise HTTPException(
            status_code=403,
            detail=f"权限不足：用户{user_id}无权限对路径{raw_path}执行{request.capability}操作"
        )
    
    # 执行对应操作
    if request.capability == "file_list":
        if not os.path.isdir(safe_path):
            raise HTTPException(status_code=400, detail="路径不是一个目录")
        return os.listdir(safe_path)
    
    elif request.capability == "file_read":
        if not os.path.isfile(safe_path):
            raise HTTPException(status_code=400, detail="路径不是一个文件")
        max_size = request.parameters.get("max_size", 1048576)
        file_size = os.path.getsize(safe_path)
        with open(safe_path, "r", encoding="utf-8", errors="ignore") as f:
            if file_size > max_size:
                return f.read(max_size) + f"\n[内容已截断，原文件大小{file_size}字节]"
            return f.read()
    
    elif request.capability == "file_write":
        overwrite = request.parameters.get("overwrite", False)
        if os.path.exists(safe_path) and not overwrite:
            raise HTTPException(status_code=400, detail="文件已存在，如需覆盖请设置overwrite=true")
        content = request.parameters.get("content", "")
        with open(safe_path, "w", encoding="utf-8") as f:
            f.write(content)
        return {"status": "success", "size": len(content), "path": raw_path}
    
    elif request.capability == "file_delete":
        if not os.path.isfile(safe_path):
            raise HTTPException(status_code=400, detail="要删除的文件不存在")
        os.remove(safe_path)
        return {"status": "success", "path": raw_path}
    
    else:
        raise HTTPException(status_code=404, detail=f"不支持的能力：{request.capability}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub) && keyMatch(r.obj, p.obj) && r.act == p.act

1
2
3
4

p, default_agent, *, file_list
p, default_agent, *, file_read
p, default_agent, *, file_write
p, default_agent, *, file_delete

1
2
3
4

mcp_services:
  - name: local_file_system
    endpoint: http://127.0.0.1:8000/mcp
    auth_type: none

1

uvicorn main:app --host 127.0.0.1 --port 8000