CLAUDE.md

1# CLAUDE.md - PostgreSQL MCP Server 开发指南
2 
3## 项目概述
4 
5PostgreSQL MCP Server 是一个基于 Model Context Protocol 的智能数据库查询服务，允许用户通过自然语言与 PostgreSQL 数据库交互。详细需求见 `specs/w5/0001-pg-mcp-prd.md`。
6 
7## 技术栈
8 
9- **Python**: 3.12+
10- **MCP SDK**: FastMCP
11- **PostgreSQL Driver**: asyncpg (异步) 或 psycopg3
12- **SQL Parser**: pglast (PostgreSQL 专用解析器)
13- **LLM**: OpenAI SDK (gpt-5.2-mini)
14- **配置管理**: pydantic-settings
15- **测试**: pytest + pytest-asyncio + pytest-cov
16 
17## 核心开发原则
18 
19### Python Best Practices
20 
21```python
22# 1. 使用类型注解 (Type Hints)
23from typing import Protocol, TypeVar, Generic
24from collections.abc import Sequence, Mapping
25 
26def query_database(sql: str, params: Mapping[str, Any] | None = None) -> list[dict[str, Any]]:
27    ...
28 
29# 2. 使用 dataclasses 或 Pydantic 定义数据模型
30from pydantic import BaseModel, Field
31 
32class QueryRequest(BaseModel):
33    question: str = Field(..., min_length=1, max_length=10000)
34    database: str | None = None
35    return_type: Literal["sql", "result"] = "result"
36 
37# 3. 使用 Enum 而非魔法字符串
38from enum import StrEnum, auto
39 
40class ErrorCode(StrEnum):
41    SUCCESS = auto()
42    SECURITY_VIOLATION = auto()
43    SQL_PARSE_ERROR = auto()
44 
45# 4. 使用 contextlib 管理资源
46from contextlib import asynccontextmanager
47 
48@asynccontextmanager
49async def get_db_connection():
50    conn = await asyncpg.connect(...)
51    try:
52        yield conn
53    finally:
54        await conn.close()
55 
56# 5. 使用 __slots__ 优化内存 (高频对象)
57class SchemaColumn:
58    __slots__ = ("name", "type", "nullable", "default", "comment")
59    ...
60```
61 
62### SOLID 原则应用
63 
64#### Single Responsibility (单一职责)
65 
66```
67src/
68├── core/
69│   ├── schema_cache.py      # 仅负责 Schema 缓存管理
70│   ├── sql_generator.py     # 仅负责 SQL 生成 (LLM 调用)
71│   ├── sql_validator.py     # 仅负责 SQL 安全验证
72│   ├── sql_executor.py      # 仅负责 SQL 执行
73│   └── result_validator.py  # 仅负责结果验证
74├── models/
75│   ├── schema.py            # Schema 相关数据模型
76│   ├── query.py             # 查询请求/响应模型
77│   └── errors.py            # 错误定义
78├── services/
79│   └── query_service.py     # 编排各个组件的服务层
80└── config/
81    └── settings.py          # 配置管理
82```
83 
84#### Open/Closed (开闭原则)
85 
86```python
87# 使用 Protocol 定义接口，便于扩展
88from typing import Protocol
89 
90class SQLGenerator(Protocol):
91    async def generate(self, question: str, schema: DatabaseSchema) -> str:
92        """Generate SQL from natural language."""
93        ...
94 
95class OpenAISQLGenerator:
96    """OpenAI 实现"""
97    async def generate(self, question: str, schema: DatabaseSchema) -> str:
98        ...
99 
100class AnthropicSQLGenerator:
101    """未来可扩展: Anthropic 实现"""
102    async def generate(self, question: str, schema: DatabaseSchema) -> str:
103        ...
104```
105 
106#### Liskov Substitution (里氏替换)
107 
108```python
109# 子类必须完全兼容父类接口
110class BaseValidator(ABC):
111    @abstractmethod
112    def validate(self, sql: str) -> ValidationResult:
113        ...
114 
115class ReadOnlyValidator(BaseValidator):
116    def validate(self, sql: str) -> ValidationResult:
117        # 返回类型和行为与父类一致
118        ...
119```
120 
121#### Interface Segregation (接口隔离)
122 
123```python
124# 细粒度接口，客户端只依赖需要的方法
125class Readable(Protocol):
126    async def read(self, key: str) -> Any: ...
127 
128class Writable(Protocol):
129    async def write(self, key: str, value: Any) -> None: ...
130 
131class SchemaCache(Readable):  # 只读缓存只实现 Readable
132    async def read(self, key: str) -> Any: ...
133```
134 
135#### Dependency Inversion (依赖反转)
136 
137```python
138# 高层模块依赖抽象，而非具体实现
139class QueryService:
140    def __init__(
141        self,
142        generator: SQLGenerator,      # 依赖抽象
143        validator: SQLValidator,       # 依赖抽象
144        executor: SQLExecutor,         # 依赖抽象
145    ):
146        self._generator = generator
147        self._validator = validator
148        self._executor = executor
149```
150 
151### DRY 原则
152 
153```python
154# 提取公共逻辑到工具函数
155# utils/sql.py
156def sanitize_identifier(name: str) -> str:
157    """安全处理 SQL 标识符"""
158    ...
159 
160# 使用装饰器消除重复的横切关注点
161from functools import wraps
162 
163def with_timeout(seconds: float):
164    def decorator(func):
165        @wraps(func)
166        async def wrapper(*args, **kwargs):
167            async with asyncio.timeout(seconds):
168                return await func(*args, **kwargs)
169        return wrapper
170    return decorator
171 
172# 使用泛型减少重复代码
173T = TypeVar("T")
174 
175class Result(Generic[T]):
176    def __init__(self, value: T | None, error: ErrorCode | None):
177        self.value = value
178        self.error = error
179```
180 
181## 代码质量要求
182 
183### 必须遵循
184 
1851. **类型完整**: 所有公开 API 必须有完整类型注解
1862. **文档字符串**: 公开类和函数必须有 docstring (Google style)
1873. **错误处理**: 使用自定义异常，不要裸露 `except`
1884. **日志脱敏**: 绝不在日志中记录密钥、密码、PII 数据
1895. **资源管理**: 使用 context manager 管理连接、文件等资源
190 
191### 代码风格
192 
193```bash
194# 使用 ruff 进行 lint 和格式化
195ruff check --fix .
196ruff format .
197 
198# pyproject.toml 配置
199[tool.ruff]
200target-version = "py312"
201line-length = 100
202 
203[tool.ruff.lint]
204select = [
205    "E", "F", "W",   # pyflakes, pycodestyle
206    "I",              # isort
207    "B", "C4",        # bugbear, comprehensions
208    "UP",             # pyupgrade
209    "SIM",            # simplify
210    "TCH",            # type-checking imports
211    "RUF",            # ruff-specific
212    "S",              # security (bandit)
213    "ASYNC",          # async best practices
214]
215 
216[tool.ruff.lint.per-file-ignores]
217"tests/**" = ["S101"]  # allow assert in tests
218```
219 
220### 安全编码
221 
222```python
223# 1. 永远不要拼接 SQL
224# BAD
225sql = f"SELECT * FROM {table_name}"
226 
227# GOOD - 使用参数化查询
228sql = "SELECT * FROM $1"
229await conn.fetch(sql, table_name)
230 
231# 2. 验证所有外部输入
232class QueryRequest(BaseModel):
233    question: str = Field(..., min_length=1, max_length=10000)
234 
235    @field_validator("question")
236    @classmethod
237    def sanitize_question(cls, v: str) -> str:
238        # 移除潜在的 prompt injection
239        return sanitize_user_input(v)
240 
241# 3. 使用 pglast 解析和验证 SQL
242from pglast import parse_sql, Node
243 
244def validate_sql(sql: str) -> bool:
245    try:
246        stmts = parse_sql(sql)
247        for stmt in stmts:
248            if not isinstance(stmt.stmt, SelectStmt):
249                raise SecurityViolationError("Only SELECT allowed")
250        return True
251    except ParseError:
252        raise SQLParseError("Invalid SQL syntax")
253```
254 
255## 测试要求
256 
257### 测试结构
258 
259```
260tests/
261├── conftest.py              # 共享 fixtures
262├── unit/
263│   ├── test_sql_validator.py
264│   ├── test_sql_generator.py
265│   └── test_schema_cache.py
266├── integration/
267│   ├── test_query_flow.py
268│   └── test_db_connection.py
269└── security/
270    ├── test_sql_injection.py
271    └── test_blocked_operations.py
272```
273 
274### 测试覆盖率要求
275 
276- **总体覆盖率**: >= 80%
277- **核心安全模块**: >= 95% (sql_validator, security checks)
278- **分支覆盖**: 必须覆盖所有安全相关分支
279 
280### 测试示例
281 
282```python
283# tests/unit/test_sql_validator.py
284import pytest
285from src.core.sql_validator import SQLValidator, SecurityViolationError
286 
287class TestSQLValidator:
288    @pytest.fixture
289    def validator(self) -> SQLValidator:
290        return SQLValidator(blocked_functions=["pg_sleep"])
291 
292    @pytest.mark.parametrize("sql", [
293        "SELECT * FROM users",
294        "SELECT COUNT(*) FROM orders WHERE date > '2024-01-01'",
295        "WITH cte AS (SELECT 1) SELECT * FROM cte",
296    ])
297    def test_valid_select_queries(self, validator: SQLValidator, sql: str):
298        assert validator.validate(sql).is_valid
299 
300    @pytest.mark.parametrize("sql,expected_error", [
301        ("DELETE FROM users", "DELETE statement not allowed"),
302        ("DROP TABLE users", "DROP statement not allowed"),
303        ("SELECT pg_sleep(100)", "Function pg_sleep is blocked"),
304        ("INSERT INTO logs VALUES (1)", "INSERT statement not allowed"),
305    ])
306    def test_blocked_operations(
307        self, validator: SQLValidator, sql: str, expected_error: str
308    ):
309        with pytest.raises(SecurityViolationError, match=expected_error):
310            validator.validate(sql)
311 
312    def test_sql_injection_attempts(self, validator: SQLValidator):
313        # 测试各种 SQL 注入变体
314        injection_attempts = [
315            "SELECT * FROM users; DROP TABLE users;--",
316            "SELECT * FROM users WHERE id = 1 OR 1=1",
317            "SELECT * FROM users UNION SELECT * FROM passwords",
318        ]
319        for sql in injection_attempts:
320            result = validator.validate(sql)
321            assert not result.allows_data_modification
322 
323# tests/integration/test_query_flow.py
324@pytest.mark.asyncio
325async def test_end_to_end_query(
326    query_service: QueryService,
327    mock_openai: MockOpenAI,
328    test_db: AsyncConnection,
329):
330    # Arrange
331    mock_openai.set_response("SELECT COUNT(*) FROM users")
332 
333    # Act
334    result = await query_service.query(
335        question="How many users are there?",
336        return_type="result",
337    )
338 
339    # Assert
340    assert result.success
341    assert result.data.row_count == 1
342    assert result.confidence >= 70
343```
344 
345### 测试运行
346 
347```bash
348# 运行所有测试
349pytest
350 
351# 运行并生成覆盖率报告
352pytest --cov=src --cov-report=html --cov-fail-under=80
353 
354# 只运行安全测试
355pytest tests/security/ -v
356 
357# 运行集成测试 (需要 PostgreSQL)
358pytest tests/integration/ --db-url="postgresql://test@localhost/test"
359```
360 
361## 性能要求
362 
363### 异步优先
364 
365```python
366# 使用 asyncpg 进行异步数据库操作
367import asyncpg
368 
369async def create_pool() -> asyncpg.Pool:
370    return await asyncpg.create_pool(
371        dsn=settings.database_url,
372        min_size=5,
373        max_size=20,
374        command_timeout=30,
375    )
376 
377# 并发执行独立操作
378async def load_all_schemas(databases: list[str]) -> dict[str, Schema]:
379    tasks = [load_schema(db) for db in databases]
380    results = await asyncio.gather(*tasks)
381    return dict(zip(databases, results))
382```
383 
384### 缓存策略
385 
386```python
387from functools import lru_cache
388from cachetools import TTLCache
389 
390# 内存缓存 Schema
391class SchemaCache:
392    def __init__(self, ttl_seconds: int = 3600):
393        self._cache: TTLCache[str, DatabaseSchema] = TTLCache(
394            maxsize=100, ttl=ttl_seconds
395        )
396 
397    async def get_schema(self, database: str) -> DatabaseSchema:
398        if database not in self._cache:
399            self._cache[database] = await self._load_schema(database)
400        return self._cache[database]
401```
402 
403### 连接池管理
404 
405```python
406# 使用连接池避免频繁建立连接
407class DatabaseManager:
408    def __init__(self):
409        self._pools: dict[str, asyncpg.Pool] = {}
410 
411    async def get_connection(self, database: str) -> asyncpg.Connection:
412        if database not in self._pools:
413            self._pools[database] = await asyncpg.create_pool(...)
414        return await self._pools[database].acquire()
415```
416 
417## 项目配置
418 
419### pyproject.toml 完整配置
420 
421```toml
422[project]
423name = "pg-mcp"
424version = "0.1.0"
425description = "PostgreSQL MCP Server for natural language queries"
426requires-python = ">=3.12"
427dependencies = [
428    "fastmcp>=2.14.1",
429    "asyncpg>=0.29.0",
430    "pglast>=6.0",
431    "openai>=1.0.0",
432    "pydantic>=2.0",
433    "pydantic-settings>=2.0",
434    "structlog>=24.0",
435]
436
437[project.optional-dependencies]
438dev = [
439    "pytest>=8.0",
440    "pytest-asyncio>=0.23",
441    "pytest-cov>=4.0",
442    "ruff>=0.4",
443    "mypy>=1.10",
444    "pre-commit>=3.0",
445]
446
447[tool.pytest.ini_options]
448asyncio_mode = "auto"
449testpaths = ["tests"]
450addopts = "-v --tb=short"
451
452[tool.mypy]
453python_version = "3.12"
454strict = true
455warn_return_any = true
456warn_unused_ignores = true
457
458[tool.coverage.run]
459source = ["src"]
460branch = true
461
462[tool.coverage.report]
463exclude_lines = [
464    "pragma: no cover",
465    "if TYPE_CHECKING:",
466    "@abstractmethod",
467]
468```
469 
470## 常用命令
471 
472```bash
473# 安装依赖
474uv sync
475 
476# 运行服务
477uv run python main.py
478 
479# 运行测试
480uv run pytest
481 
482# 类型检查
483uv run mypy src
484 
485# Lint 和格式化
486uv run ruff check --fix .
487uv run ruff format .
488 
489# 生成覆盖率报告
490uv run pytest --cov=src --cov-report=html
491```
492 
493## 错误处理模式
494 
495```python
496# 定义项目异常层次
497class PgMcpError(Exception):
498    """Base exception for pg-mcp"""
499    def __init__(self, message: str, code: ErrorCode):
500        super().__init__(message)
501        self.code = code
502 
503class SecurityViolationError(PgMcpError):
504    def __init__(self, message: str):
505        super().__init__(message, ErrorCode.SECURITY_VIOLATION)
506 
507class SQLParseError(PgMcpError):
508    def __init__(self, message: str):
509        super().__init__(message, ErrorCode.SQL_PARSE_ERROR)
510 
511# 统一错误处理
512async def handle_query(request: QueryRequest) -> QueryResponse:
513    try:
514        return await _process_query(request)
515    except SecurityViolationError as e:
516        return QueryResponse(
517            success=False,
518            error=ErrorInfo(code=e.code, message=str(e)),
519        )
520    except PgMcpError as e:
521        logger.warning("Query failed", error=str(e), code=e.code)
522        return QueryResponse(success=False, error=ErrorInfo(code=e.code, message=str(e)))
523    except Exception:
524        logger.exception("Unexpected error")
525        return QueryResponse(
526            success=False,
527            error=ErrorInfo(code=ErrorCode.INTERNAL_ERROR, message="Internal error"),
528        )
529```
530 
531## Git 提交规范
532 
533```
534feat: 新功能
535fix: Bug 修复
536docs: 文档更新
537refactor: 重构 (不改变功能)
538test: 测试相关
539perf: 性能优化
540security: 安全相关修复
541```
542 
543## 检查清单
544 
545在提交 PR 前确保:
546 
547- [ ] 所有测试通过 (`pytest`)
548- [ ] 类型检查通过 (`mypy src`)
549- [ ] Lint 检查通过 (`ruff check .`)
550- [ ] 安全测试覆盖新增代码路径
551- [ ] 敏感信息未暴露在日志中
552- [ ] 文档已更新 (如适用)
553