使用说明

高效代码库分析MCP服务器架构计划

概述

一种MCP服务器，提供智能代码库分析，无需一次将所有内容加载到内存中。代码库托管在GitHub上，并在本地与高效的基于git的更新同步。

核心架构

1.存储库同步层

初始克隆

def initialize_repository(repo_url, local_path):
    """
    Clone repository with optimizations
    """
    # Shallow clone to save bandwidth and storage
    git.clone(
        repo_url, 
        local_path,
        depth=1,  # Only latest commit initially
        single_branch=True,  # Only main branch
        filter='blob:none'  # Blobless clone - fetch files on demand
    )
    
    # Store metadata
    store_repo_metadata({
        'url': repo_url,
        'local_path': local_path,
        'current_commit': get_current_commit_hash(),
        'last_synced': timestamp()
    })

增量更新策略

def sync_repository():
    """
    Pull only changes since last sync
    """
    current_commit = get_stored_commit_hash()
    
    # Fetch changes
    git.fetch(origin='origin', depth=1)
    
    # Get diff between commits
    changed_files = git.diff(
        current_commit,
        'origin/main',
        name_only=True
    )
    
    # Categorize changes
    changes = {
        'added': [],
        'modified': [],
        'deleted': [],
        'renamed': []
    }
    
    for file_info in git.diff_tree(current_commit, 'origin/main'):
        status = file_info.status
        if status == 'A':
            changes['added'].append(file_info.path)
        elif status == 'M':
            changes['modified'].append(file_info.path)
        elif status == 'D':
            changes['deleted'].append(file_info.path)
        elif status == 'R':
            changes['renamed'].append({
                'old': file_info.old_path,
                'new': file_info.new_path
            })
    
    # Pull the actual changes
    git.pull(origin='origin', branch='main')
    
    # Update indexes incrementally
    process_changes(changes)
    
    # Update commit hash
    update_stored_commit_hash(get_current_commit_hash())
    
    return changes

智能同步调度

按需：当用户进行查询时，检查是否需要同步
周期性的：每N分钟进行一次背景同步（可配置）
Webhook已触发：如果配置了GitHub webhook，则同步推送事件
手册：公开用于显式同步请求的工具

2.索引阶段（一次性+增量更新）

Git感知文件系统爬虫

目的：构建代码库的轻量级索引
策略:
- 使用git ls树进行文件发现（比文件系统扫描更快）
- 从git中提取元数据（提交哈希、作者、最后修改时间）
- 尊重 .gitignore 通过git自动
- 使用git责备来跟踪文件历史记录
- 以pgvector扩展名存储在PostgreSQL数据库中

符号提取器

目的：提取高级代码结构而不进行完全解析
索引什么:
- 函数/方法签名
- 类定义
- 进出口报表
- 注释和文档字符串（仅第一行）
- 文件级元数据
工具：使用特定语言的解析器（Tree sitter用于多语言支持）

存储：结构化索引，包括：

{
  file_path: string,
  symbols: [{name, type, line_start, line_end, signature}],
  imports: [string],
  exports: [string],
  hash: string  // for change detection
}

依赖关系图生成器

目的：映射文件之间的关系
数据:
- 导入/要求关系
- 调用图（轻量级，符号级）
- 继承体系
存储：图形数据库或邻接列表

基于pgvector的语义分块

目的：创建可搜索的上下文块
策略:
- 按逻辑单元（函数、类、模块）分组
- 包括周围环境（导入、相关定义）
- 生成语义搜索的嵌入
- 使用pgvector扩展名存储在PostgreSQL中
- 存储块元数据：文件、行号、父符号、git提交哈希

矢量存储模式:

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Code chunks table with embeddings
CREATE TABLE code_chunks (
    id SERIAL PRIMARY KEY,
    file_id INTEGER REFERENCES files(id),
    chunk_text TEXT,
    chunk_type TEXT,  -- 'function', 'class', 'module', etc.
    symbol_name TEXT,
    line_start INTEGER,
    line_end INTEGER,
    embedding vector(384),  -- Dimension depends on model
    commit_hash TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Create HNSW index for fast vector search
CREATE INDEX ON code_chunks 
USING hnsw (embedding vector_cosine_ops);

-- Create GiST index for metadata filtering
CREATE INDEX idx_chunks_file ON code_chunks(file_id);
CREATE INDEX idx_chunks_symbol ON code_chunks(symbol_name);
CREATE INDEX idx_chunks_type ON code_chunks(chunk_type);

2.查询处理流水线

查询分析层

User Query → Intent Classification → Strategy Selection

意图类型:

find_definition：定位定义某物的位置
find_usage：查找符号的所有用法
explain_function：了解特定代码
trace_flow：遵循执行路径
find_similar：语义代码搜索
architectural：高层结构问题

战略路由器

根据意图，选择适当的检索策略：

策略1：基于符号的查找 （快速）

For：“函数X在哪里定义？”
流程：直接索引查找→ 返回文件+行号
加载：仅加载相关文件部分

策略2：依赖遍历 （中等）

For：“此模块依赖于什么？”
流程：图形查询→ 返回连接节点
加载：仅依赖元数据

策略3：使用pgvector进行语义搜索 （中速）

For：“查找处理身份验证的代码”
流程：
1. 使用与索引相同的模型嵌入查询
2. 基于pgvector余弦相似度的向量相似度搜索
3. 使用元数据过滤器以获得更好的结果
4. 按相关性+新近度排名
5. 返回带有上下文的前k个块

SQL查询：

SELECT 
    c.id,
    c.chunk_text,
    c.symbol_name,
    c.line_start,
    c.line_end,
    f.path,
    1 - (c.embedding <=> $1::vector) as similarity
FROM code_chunks c
JOIN files f ON c.file_id = f.id
WHERE 
    c.chunk_type = ANY($2)  -- Optional: filter by type
    AND 1 - (c.embedding <=> $1::vector) > 0.7  -- Similarity threshold
ORDER BY c.embedding <=> $1::vector
LIMIT 10;

加载：仅匹配块+周围上下文

策略4：深度分析 （慢速，按需）

For：“解释支付流程是如何工作的”
流程：
1. 使用语义搜索查找入口点
2. 遍历依赖关系图
3. 仅加载相关文件
4. 为法学硕士构建专注的背景
5. 增量流分析

3.上下文组装

智能上下文生成器

def build_context(query, max_tokens=8000):
    relevant_symbols = find_relevant(query)
    context = []
    token_count = 0
    
    # Priority order
    for symbol in rank_by_relevance(relevant_symbols):
        chunk = load_chunk(symbol)
        if token_count + len(chunk) < max_tokens:
            context.append(chunk)
            token_count += len(chunk)
        else:
            break
    
    return context

排名因素:

查询的语义相似性
文件修改的最近时间
导入深度（首选高级模块）
符号类型（定义优先于使用）
用户最近的上下文（如果可用）

4.MCP服务器工具

工具设计

// Tool 1: Search Codebase
{
  name: "search_code",
  description: "Search for code using natural language or symbols",
  parameters: {
    query: "authentication logic",
    search_type: "semantic" | "symbol" | "text",
    max_results: 5,
    file_filter: "*.ts",  // Optional glob pattern
    similarity_threshold: 0.7  // For semantic search
  }
}

// Tool 2: Get Definition
{
  name: "get_definition",
  description: "Get the definition of a symbol",
  parameters: {
    symbol_name: "handlePayment",
    file_hint: "optional/path/hint.ts"
  }
}

// Tool 3: Find Usages
{
  name: "find_usages",
  description: "Find all usages of a symbol",
  parameters: {
    symbol_name: "UserModel",
    include_context: true
  }
}

// Tool 4: Explain Code
{
  name: "explain_code",
  description: "Get detailed explanation of code section",
  parameters: {
    file_path: "src/utils/auth.ts",
    line_start: 45,
    line_end: 80
  }
}

// Tool 5: Trace Execution
{
  name: "trace_execution",
  description: "Follow execution path from entry point",
  parameters: {
    entry_point: "handleRequest",
    max_depth: 3
  }
}

// Tool 6: Get Architecture
{
  name: "get_architecture",
  description: "Get high-level architecture overview",
  parameters: {
    focus_area: "backend" | "frontend" | "full"
  }
}

// Tool 7: Sync Repository
{
  name: "sync_repository",
  description: "Pull latest changes from GitHub",
  parameters: {
    force: false  // Force full re-index if true
  }
}

// Tool 8: Get Repository Status
{
  name: "get_repo_status",
  description: "Get current repository sync status",
  parameters: {}
  // Returns: current commit, last sync time, pending changes
}

5.缓存策略

多级缓存

L1：内存缓存

最近访问的块
频繁查询及其结果
为常见查询嵌入缓存
符号查找缓存
TTL：5分钟
大小：最大100MB

L2：磁盘缓存

解析后的文件内容（按提交哈希+文件路径键入）
中间分析结果
预先计算的依赖路径
TTL：直到文件在git中更改
地点： ~/.cache/mcp-codebase/

L3：带pgvector的PostgreSQL

持久符号索引
依赖图
文件元数据
矢量嵌入
更新：关于git diff检测

缓存无效策略:

def invalidate_cache(file_path):
    """
    Invalidate caches when file changes detected via git
    """
    # L1: Clear in-memory cache for this file
    memory_cache.delete(f"file:{file_path}")
    memory_cache.delete_pattern(f"query:*:{file_path}")
    
    # L2: Remove disk cache entries
    disk_cache.delete(f"{current_commit}:{file_path}")
    
    # L3: Updated via process_file_update()
    # Embeddings are versioned by commit_hash

6.增量更新（基于Git）

Git Diff处理器

def process_changes(changes):
    """
    Process git diff and update indexes incrementally
    """
    # Handle deletions first
    for deleted_file in changes['deleted']:
        file_id = get_file_id(deleted_file)
        if file_id:
            # Delete from database
            delete_symbols(file_id)
            delete_chunks(file_id)
            delete_dependencies(file_id)
            delete_file_record(file_id)
    
    # Handle renames
    for rename in changes['renamed']:
        update_file_path(rename['old'], rename['new'])
    
    # Handle additions and modifications
    files_to_process = changes['added'] + changes['modified']
    
    for file_path in files_to_process:
        # Check if file should be indexed (not binary, not too large)
        if should_index_file(file_path):
            process_file_update(file_path)

def process_file_update(file_path):
    """
    Update index for a single file
    """
    # Compute hash
    content = read_file(file_path)
    new_hash = compute_hash(content)
    
    file_record = get_file_record(file_path)
    
    if file_record and file_record.hash == new_hash:
        # No actual changes, skip
        return
    
    # Extract symbols
    symbols = extract_symbols(content, file_path)
    
    # Update or insert file record
    file_id = upsert_file(file_path, new_hash)
    
    # Delete old symbols and chunks
    delete_symbols(file_id)
    delete_chunks(file_id)
    
    # Insert new symbols
    for symbol in symbols:
        insert_symbol(file_id, symbol)
    
    # Generate chunks and embeddings
    chunks = create_chunks(content, symbols, file_path)
    
    # Batch embed chunks
    embeddings = generate_embeddings([chunk.text for chunk in chunks])
    
    # Insert chunks with embeddings into PostgreSQL
    for chunk, embedding in zip(chunks, embeddings):
        insert_chunk_with_embedding(
            file_id=file_id,
            chunk=chunk,
            embedding=embedding,
            commit_hash=get_current_commit_hash()
        )
    
    # Update dependencies
    update_dependencies(file_id, content)
    
    # Invalidate related caches
    invalidate_cache(file_path)

def insert_chunk_with_embedding(file_id, chunk, embedding, commit_hash):
    """
    Insert chunk with pgvector embedding
    """
    execute_sql("""
        INSERT INTO code_chunks 
        (file_id, chunk_text, chunk_type, symbol_name, line_start, line_end, embedding, commit_hash)
        VALUES ($1, $2, $3, $4, $5, $6, $7::vector, $8)
    """, [
        file_id,
        chunk.text,
        chunk.type,
        chunk.symbol_name,
        chunk.line_start,
        chunk.line_end,
        embedding.tolist(),  # Convert numpy array to list
        commit_hash
    ])

Webhook集成（可选）

# FastAPI endpoint for GitHub webhooks
@app.post("/webhook/github")
async def github_webhook(payload: dict, signature: str):
    """
    Handle GitHub push events
    """
    # Verify webhook signature
    if not verify_signature(payload, signature):
        raise HTTPException(401)
    
    if payload['ref'] == 'refs/heads/main':
        # Trigger sync in background
        background_tasks.add_task(sync_repository)
        
    return {"status": "ok"}

7.性能优化

延迟加载

仅在需要时加载文件内容
仅对活动查询解析语法
按需生成嵌入（带缓存）

并行处理

同时索引多个文件
批量嵌入生成
并行依赖解析

智能采样

对于大文件，请分段分析
在完整内容之前提供文件摘要
渐进式细节加载

8.实现栈

核心技术:

MCP框架：模型上下文协议SDK
数据库：PostgreSQL 15+，带pgvector扩展
Git集成：GitPython或libgit2（pygit2）
解析：树保姆（多语言）
嵌入：句子转换器（全MiniLM-L6-v2或全mpnet-base-v2）
语言：带asyncio的Python 3.10+
API框架：FastAPI（用于webhook端点）
后台任务：Celery或异步任务

数据库模式:

-- Repository metadata
CREATE TABLE repositories (
    id SERIAL PRIMARY KEY,
    url TEXT UNIQUE NOT NULL,
    local_path TEXT NOT NULL,
    current_commit TEXT,
    last_synced TIMESTAMP,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Files table
CREATE TABLE files (
    id SERIAL PRIMARY KEY,
    repo_id INTEGER REFERENCES repositories(id),
    path TEXT NOT NULL,
    hash TEXT,
    language TEXT,
    size_bytes INTEGER,
    last_modified TIMESTAMP,
    commit_hash TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(repo_id, path)
);

CREATE INDEX idx_files_repo ON files(repo_id);
CREATE INDEX idx_files_path ON files(path);
CREATE INDEX idx_files_commit ON files(commit_hash);

-- Symbols table
CREATE TABLE symbols (
    id SERIAL PRIMARY KEY,
    file_id INTEGER REFERENCES files(id) ON DELETE CASCADE,
    name TEXT NOT NULL,
    type TEXT,  -- 'function', 'class', 'variable', etc.
    line_start INTEGER,
    line_end INTEGER,
    signature TEXT,
    docstring TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_symbols_file ON symbols(file_id);
CREATE INDEX idx_symbols_name ON symbols(name);
CREATE INDEX idx_symbols_type ON symbols(type);

-- Dependencies table
CREATE TABLE dependencies (
    id SERIAL PRIMARY KEY,
    from_file_id INTEGER REFERENCES files(id) ON DELETE CASCADE,
    to_file_id INTEGER REFERENCES files(id) ON DELETE CASCADE,
    import_type TEXT,  -- 'import', 'require', 'include'
    symbol_name TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(from_file_id, to_file_id, symbol_name)
);

CREATE INDEX idx_deps_from ON dependencies(from_file_id);
CREATE INDEX idx_deps_to ON dependencies(to_file_id);

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Code chunks table with embeddings
CREATE TABLE code_chunks (
    id SERIAL PRIMARY KEY,
    file_id INTEGER REFERENCES files(id) ON DELETE CASCADE,
    chunk_text TEXT NOT NULL,
    chunk_type TEXT,  -- 'function', 'class', 'module', 'comment'
    symbol_name TEXT,
    line_start INTEGER,
    line_end INTEGER,
    embedding vector(384),  -- Adjust dimension based on model
    commit_hash TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- HNSW index for fast approximate nearest neighbor search
CREATE INDEX idx_chunks_embedding ON code_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Additional indexes for filtering
CREATE INDEX idx_chunks_file ON code_chunks(file_id);
CREATE INDEX idx_chunks_symbol ON code_chunks(symbol_name);
CREATE INDEX idx_chunks_type ON code_chunks(chunk_type);
CREATE INDEX idx_chunks_commit ON code_chunks(commit_hash);

环境配置:

# config.yaml
repository:
  url: "https://github.com/username/repo.git"
  local_path: "./repos/repo"
  branch: "main"
  sync_interval_minutes: 15
  
database:
  host: "localhost"
  port: 5432
  database: "codebase_mcp"
  user: "mcp_user"
  password: "${DB_PASSWORD}"

embedding:
  model: "sentence-transformers/all-MiniLM-L6-v2"
  dimension: 384
  batch_size: 32
  device: "cuda"  # or "cpu"

indexing:
  max_file_size_kb: 1024
  supported_extensions: [".py", ".js", ".ts", ".java", ".go", ".rs"]
  chunk_size_lines: 50
  chunk_overlap_lines: 10

cache:
  memory_size_mb: 100
  disk_path: "~/.cache/mcp-codebase"

webhooks:
  enabled: true
  secret: "${GITHUB_WEBHOOK_SECRET}"
  port: 8080

查询流示例

查询：“身份验证系统是如何工作的？”

同步检查：检查仓库是否需要同步（上次同步>15分钟前）
- 如果是：运行 sync_repository() 在后台
- 继续使用当前索引数据
查询分析：意图= architectural + explain
符号搜索：查询PostgreSQL中的“auth*”符号→ 15 火柴

语义搜索:

嵌入查询：“身份验证系统”→ 矢量（384）
查询pgvector：

SELECT c.*, f.path, 1 - (c.embedding <=> $1::vector) as similarity
FROM code_chunks c
JOIN files f ON c.file_id = f.id
WHERE 1 - (c.embedding <=> $1::vector) > 0.7
ORDER BY c.embedding <=> $1::vector
LIMIT 5;

获取前5个相关块

依赖遍历：从依赖关系图中获取身份验证模块依赖关系
上下文组装:
- 从git加载主认证模块（2KB）
- 加载密钥中间件（1.5KB）
- 加载与身份验证相关的配置（0.5KB）
- 总计：4KB，8KB预算
响应：提供结构化的解释，包括：
- 带有GitHub URL的文件引用
- 行号
- 提交哈希用于版本跟踪
- 定义链接

Git特定优化

1.金发克隆

git clone --filter=blob:none --depth=1 <repo-url>

最初只下载树和提交对象
按需获取文件内容
为大型存储库节省约70%的初始克隆时间

2.备件检查（可选）

对于monorepos，只签出相关目录：

def setup_sparse_checkout(paths):
    """
    Only checkout specified paths
    """
    git.config('core.sparseCheckout', 'true')
    with open('.git/info/sparse-checkout', 'w') as f:
        for path in paths:
            f.write(f"{path}\n")
    git.read_tree('-mu', 'HEAD')

3.Git对象数据库查询

不要从磁盘读取文件，直接查询git：

def get_file_at_commit(file_path, commit_hash):
    """
    Get file content without checkout
    """
    return git.show(f"{commit_hash}:{file_path}")

4.高效的差分处理

def get_changed_files_efficient(from_commit, to_commit):
    """
    Get changed files without full checkout
    """
    return git.diff_tree(
        '--no-commit-id',
        '--name-status',
        '-r',
        from_commit,
        to_commit
    )

可扩展性考虑因素

小代码库（<1000个文件）:
- 全嵌入索引
- 每5分钟同步一次
- 快速查询（<500ms）
中等（1000-10000个文件）:
- 选择性嵌入（优先处理频繁访问的文件）
- 每15分钟同步一次
- 使用pgvector的HNSW索引进行快速搜索
- 索引过程中的并行处理
大型（>10000个文件）:
- 分层索引（首先进行模块级摘要）
- 智能同步（仅限活动分支）
- GPU批量嵌入生成
- 考虑多个工作进程
- 按存储库/模块使用PostgreSQL分区
- 浅深度增量git获取

PostgreSQL性能调优:

-- Tune HNSW index parameters for your dataset
CREATE INDEX idx_chunks_embedding ON code_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (
    m = 16,              -- Increase for better recall (16-64)
    ef_construction = 64  -- Increase for better index quality (64-200)
);

-- Query-time tuning
SET hnsw.ef_search = 100;  -- Higher = better accuracy, slower

-- Connection pooling
-- Use pgBouncer or SQLAlchemy pool for concurrent requests

嵌入生成优化:

def batch_embed_chunks(chunks, batch_size=32):
    """
    Generate embeddings in batches for efficiency
    """
    model = SentenceTransformer('all-MiniLM-L6-v2')
    model.to('cuda' if torch.cuda.is_available() else 'cpu')
    
    all_embeddings = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i+batch_size]
        embeddings = model.encode(
            batch,
            convert_to_numpy=True,
            show_progress_bar=False
        )
        all_embeddings.extend(embeddings)
    
    return all_embeddings

成功指标

查询响应时间：90%的查询<2s（符号查找<200ms）
初始克隆+索引时间：10k文件<5分钟
增量同步时间：典型PR更改（10-50个文件）小于30秒
索引大小：小于代码库大小的10%
准确率：前三名语义搜索结果的相关性>85%
PostgreSQL存储：约2-3x代码库大小（包括嵌入）
同步频率：每15分钟一次（可配置）
内存使用率：驻留内存<500MB（无GPU）

部署架构

┌─────────────────────┐
│   GitHub Repo       │
│   (Remote)          │
└──────────┬──────────┘
           │
           │ git fetch/pull
           │ (incremental)
           ▼
┌─────────────────────┐      ┌──────────────────┐
│   Local Git Clone   │      │  PostgreSQL +    │
│   (Shallow/Blobless)│◄────►│  pgvector        │
└──────────┬──────────┘      └────────┬─────────┘
           │                          │
           │                          │
           ▼                          ▼
┌─────────────────────┐      ┌──────────────────┐
│  MCP Server         │◄────►│  Embedding Model │
│  (FastAPI/Python)   │      │  (sentence-trans)│
└──────────┬──────────┘      └──────────────────┘
           │
           │ MCP Protocol
           ▼
┌─────────────────────┐
│   Claude Desktop    │
│   or API Client     │
└─────────────────────┘