使用说明

MCP企业堆栈

一个三微服务系统，通过模型上下文协议（MCP）公开PostgreSQL数据仓库。构建用于在现实的多团队数据环境中演示企业级SQL访问控制、语义列映射和查询安全。

它的作用

大多数数据仓库都有同样的问题：同一概念在表中有不同的列名。财务呼唤收入 amount市场营销称之为 spend销售人员称之为 deal_value任何试图跨团队查询的工具都必须硬编码这些映射或猜错。

这个堆栈通过概念注册表解决了这个问题——在运行时用语义概念标记列，任何工具都可以自动发现JOIN路径。最重要的是，每个查询都通过Postgres行级安全机制运行，因此无论编写什么SQL，分析师都只能看到自己的行。

建筑学

┌─────────────────────────────────────────────────────┐
│                    MCP Client                        │
│              (any MCP-compatible tool)               │
└──────────────────────┬──────────────────────────────┘
                       │  MCP protocol (stdio)
                       ▼
┌─────────────────────────────────────────────────────┐
│                  mcp-server/                         │
│  FastMCP — 13 tools: query, explain, suggest_join,  │
│  compare_metric, query_async, cache_stats, ...       │
│  • _guard(): app-layer access check                  │
│  • _validate_sql(): AST-based SQL allowlist          │
│  • _set_session_ctx(): stamps RLS identity           │
└───────────┬─────────────────────┬────────────────────┘
            │ HTTP                │ HTTP
            ▼                     ▼
┌───────────────────┐  ┌──────────────────────────────┐
│  access-control/  │  │       semantic-mapper/        │
│  FastAPI :8001    │  │       FastAPI :8000            │
│                   │  │                               │
│  API key → user   │  │  concept registry             │
│  identity + role  │  │  column tagging               │
│  + allowed tables │  │  JOIN path discovery          │
└───────────────────┘  └──────────────────────────────┘
            │                     │
            └──────────┬──────────┘
                       ▼
        ┌──────────────────────────┐
        │      PostgreSQL          │
        │  Row-Level Security on   │
        │  all 4 business tables   │
        │  + query_cache           │
        │  + query_audit_log       │
        │  + async_tasks           │
        └──────────────────────────┘

主要特点

基于角色的访问控制

通过访问控制服务进行API密钥验证
四个角色： analyst, manager, director, admin
团队范围的表访问：销售团队无法查询营销表
_guard() 在查询到达数据库之前阻止未经授权的表访问

PostgreSQL行级安全

每个业务表上的RLS策略
会话变量(app.user_role, app.user_team, app.user_name)按查询设置
分析人士只看到自己的观点；管理者看到他们的团队；董事看到所有
事务局部变量防止连接之间的会话状态泄漏

语义列映射器

6个语义概念的中央注册表： customer, date, geography, money, person, product
通过REST API标记具有概念的任何列-不需要架构迁移
GET /join-path?from_table=X&to_table=Y 返回两个表之间的所有有效JOIN条件
自我加入 column_tags 使发现查询成为单个SQL语句

SQL安全

基于AST的验证 sqlglot 在每次查询之前
仅 SELECT CTE(WITH)allowed--DDL、DML和函数调用被阻止
阻止列表捕获字符串匹配缺失的模式限定函数名和大小写变量函数名

查询基础结构

Postgres中的结果缓存(query_cache 表）——由规范化的SQL+用户标识键控，因此RLS上下文被烘焙到缓存键中
完整审计日志(query_audit_log)--每个查询都印有用户、团队、行数、持续时间
使用轮询异步执行查询(query_async + get_task_result)
NullPool --应用程序中没有连接池；在外部使用RDS代理进行多实例部署

项目结构

MCP_Project/
├── mcp-server/
│   ├── server.py          # FastMCP server — all 13 tool definitions
│   ├── db.py              # SQLAlchemy layer: validate, cache, RLS, audit
│   ├── pyproject.toml
│   └── .env.example
│
├── access-control/
│   ├── main.py            # FastAPI: /me, /check, /users
│   ├── db.py              # User store, API key lookup, table permission logic
│   ├── seed_users.py      # Seeds 9 demo users with generated API keys
│   ├── pyproject.toml
│   └── .env.example
│
├── semantic-mapper/
│   ├── main.py            # FastAPI: /concepts, /tags, /join-path, tag/untag
│   ├── db.py              # Concept registry + JOIN discovery query
│   ├── seed_registry.py   # Seeds 6 concepts and column tags for demo tables
│   ├── pyproject.toml
│   └── .env.example
│
├── start.sh               # Starts both microservices
└── docs/
    └── DEMO.md            # Demo walkthrough

设置

先决条件

PostgreSQL在本地运行
紫外线包管理器
名为的Postgres数据库 mcp_project

1.创建数据库和用户

psql postgres -c "CREATE DATABASE mcp_project;"
psql postgres -c "CREATE USER app_user;"
psql mcp_project -c "GRANT ALL ON SCHEMA public TO app_user;"

2.配置环境变量

复制示例文件并设置数据库URL：

cp mcp-server/.env.example       mcp-server/.env
cp access-control/.env.example   access-control/.env
cp semantic-mapper/.env.example  semantic-mapper/.env

编辑每个 .env 以匹配您的Postgres连接字符串。

3.安装依赖项

cd mcp-server      && uv sync && cd ..
cd access-control  && uv sync && cd ..
cd semantic-mapper && uv sync && cd ..

4.启动微服务

./start.sh

服务将在以下地点提供：

语义映射器： http://localhost:8000 （斯瓦格： /docs)
访问控制： http://localhost:8001 （斯瓦格： /docs)

5.为数据库添加种子

# Seed users and generate API keys
cd access-control && uv run python seed_users.py

# Seed semantic concept registry
cd ../semantic-mapper && uv run python seed_registry.py

# Seed business data + RLS policies
cd ../mcp-server
uv run python seed.py
uv run python migrate_rls.py
uv run python patch_analyst_rows.py  # sets rep_name for RLS filtering

6.获取API密钥

# Get the admin key first — it's printed during seed_users.py
# Then list all users:
curl -s http://localhost:8001/users -H "X-Api-Key: <admin_key>"

7.配置MCP服务器

集 MCP_USER_API_KEY 在 mcp-server/.env 指向要运行的用户的API密钥。将MCP客户端指向 mcp-server/server.py.

演示流程

最有效的演示显示了相同的SQL查询根据询问者返回不同的行数：

集 MCP_USER_API_KEY 到管理员密钥→ query("SELECT count(*) FROM sales_pipeline") → 160
转到Grace Lin（销售分析师）→ 相同的查询→ 10
转到亨利·布鲁克斯（营销分析师）→ describe_table("sales_pipeline") → 权限错误

看 docs/DEMO.md 了解完整的演示演练。

技术栈

FastMCP --MCP服务器框架
快速API --REST微服务
SQLAlchemy --数据库层（NullPool）
sqlglot --用于验证的SQL AST解析器
PostgreSQL --RLS、会话变量、JSONB缓存
紫外线 --包管理

RLS策略结构

每个业务表都有一个策略，如：

CREATE POLICY rls_sales_pipeline ON sales_pipeline
  USING (
    current_setting('app.user_role', true) IN ('director', 'admin')
    OR (
      current_setting('app.user_role', true) = 'manager'
      AND current_setting('app.user_team', true) = 'sales'
    )
    OR (
      current_setting('app.user_role', true) = 'analyst'
      AND current_setting('app.user_team', true) = 'sales'
      AND rep_name = current_setting('app.user_name', true)
    )
  );

会话变量在每次查询之前都在本地设置事务——它们不能跨连接持久化。

扩展堆栈

添加新表：

在Postgres中创建表并设置种子
添加到 TEAM_TABLES 在 access-control/db.py
通过以下方式标记其列 POST /columns/{table}/{column}/tag
添加RLS策略 migrate_rls.py 或手动

添加新用户：

curl -X POST http://localhost:8001/users \
  -H "X-Api-Key: <admin_key>" \
  -H "Content-Type: application/json" \
  -d '{"username": "new_user", "full_name": "New User", "team": "sales", "role": "analyst"}'

添加新的语义概念：

curl -X POST http://localhost:8000/columns/my_table/my_column/tag \
  -H "Content-Type: application/json" \
  -d '{"concept": "revenue"}'