企业级落地方案：Docker 部署 CodeGraph 多项目统一 MCP 网关，附源码

在微服务架构逐渐成为主流的今天，企业往往面临多个项目并行开发、代码仓库分散、代码依赖关系复杂等问题。为了提升研发效能，我们需要一个统一的代码图（CodeGraph）管理与模型上下文协议（MCP）网关，来聚合代码元数据、提供智能代码查询与上下文推送能力。本文将从实战角度出发，详细讲解如何使用 Docker 部署一个多项目统一的 MCP 网关，并附上完整的可运行源码示例。## 背景与需求在企业级开发中，通常存在以下痛点：- 多个项目（如 frontend、backend、mobile）使用不同的语言和技术栈，代码分析工具难以统一。- 代码变更频繁，跨项目调用链和依赖关系难以追踪。- 缺乏一个统一的网关来聚合代码元数据，供 AI 助手或内部工具查询上下文。我们的目标是：构建一个基于 CodeGraph 的 MCP 网关，能够：1. 扫描多个 Git 仓库，生成统一的代码图。2. 通过 REST API 提供代码查询、依赖分析、上下文推送。3. 使用 Docker 容器化部署，支持水平扩展。## 系统架构设计我们采用以下组件：- CodeGraph Engine：解析代码生成图数据结构（基于 tree-sitter 或类似工具）。- MCP Gateway：基于 FastAPI 的 REST 网关，聚合多项目元数据，提供统一查询接口。- PostgreSQL：存储代码图元数据（节点、边、属性）。- Redis：缓存热点查询结果，提升性能。- Docker Compose：编排所有服务。架构图如下（文字描述）：[Git Repos] --> [CodeGraph Engine] --> [PostgreSQL] |[MCP Gateway] <--> [Redis] <--> [PostgreSQL] |[外部客户端/ AI助手] --> [MCP Gateway]## 环境准备确保你已经安装：- Docker 20.10± Docker Compose 1.29± Git## 实战步骤：构建与部署### 第一步：项目结构创建以下目录结构：codegraph-mcp-gateway/├── docker-compose.yml├── codegraph-engine/│ ├── Dockerfile│ ├── requirements.txt│ └── engine.py├── mcp-gateway/│ ├── Dockerfile│ ├── requirements.txt│ ├── main.py│ └── models.py├── init.sql└── repos/ ├── project-a/ └── project-b/### 第二步：编写 CodeGraph 引擎（core）codegraph-engine/engine.py：扫描仓库生成代码图。python# codegraph-engine/engine.pyimport osimport jsonimport hashlibfrom pathlib import Pathfrom typing import List, Dict# 模拟的代码解析函数（实际可用 tree-sitter 或 ast）def parse_code_file(file_path: str) -> List[Dict]: """ 解析单个代码文件，提取函数、类、导入关系。 返回节点列表：[{type: 'function', name: 'foo', file: '...', line: 10}, ...] """ nodes = [] try: with open(file_path, 'r', encoding='utf-8') as f: lines = f.readlines() for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith('def '): func_name = stripped.split('(')[0].replace('def ', '').strip() nodes.append({ 'type': 'function', 'name': func_name, 'file': file_path, 'line': i + 1, 'hash': hashlib.md5(f"{file_path}:{func_name}".encode()).hexdigest() }) elif stripped.startswith('import ') or stripped.startswith('from '): nodes.append({ 'type': 'import', 'name': stripped, 'file': file_path, 'line': i + 1, 'hash': hashlib.md5(f"{file_path}:{i}".encode()).hexdigest() }) except Exception as e: print(f"Error parsing {file_path}: {e}") return nodesdef scan_repo(repo_path: str) -> Dict: """ 扫描整个仓库，生成代码图数据结构。 返回：{'nodes': [...], 'edges': [...]} """ graph = {'nodes': [], 'edges': []} repo_path = Path(repo_path) for file_path in repo_path.rglob('*.py'): # 支持 Python 项目，可扩展 nodes = parse_code_file(str(file_path)) graph['nodes'].extend(nodes) # 简单边示例：函数调用关系（此处简化） for node in nodes: if node['type'] == 'function': graph['edges'].append({ 'source': node['hash'], 'target': 'root', 'relation': 'defined_in' }) # 去重 graph['nodes'] = list({n['hash']: n for n in graph['nodes']}.values()) return graphif __name__ == '__main__': import sys if len(sys.argv) != 2: print("Usage: engine.py <repo_path>") sys.exit(1) repo_path = sys.argv[1] graph = scan_repo(repo_path) print(json.dumps(graph, indent=2))### 第三步：编写 MCP 网关（FastAPI）mcp-gateway/main.py：提供统一查询 API。python# mcp-gateway/main.pyfrom fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import List, Optionalimport asyncpgimport redis.asyncio as redisimport jsonapp = FastAPI(title="CodeGraph MCP Gateway")# 数据模型class Node(BaseModel): hash: str type: str name: str file: str line: intclass Edge(BaseModel): source: str target: str relation: strclass GraphQuery(BaseModel): project: Optional[str] = None node_type: Optional[str] = None name_contains: Optional[str] = None# 连接池（实际应配置环境变量）DB_URL = "postgresql://user:password@db:5432/codegraph"REDIS_URL = "redis://redis:6379/0"@app.on_event("startup")async def startup(): app.state.pg = await asyncpg.create_pool(DB_URL) app.state.redis = await redis.from_url(REDIS_URL)@app.post("/query")async def query_graph(query: GraphQuery): """ 查询代码图，支持按项目、节点类型、名称过滤。 """ # 尝试从缓存读取 cache_key = f"graph:{query.project}:{query.node_type}:{query.name_contains}" cached = await app.state.redis.get(cache_key) if cached: return json.loads(cached) # 构建 SQL 查询 sql = "SELECT * FROM nodes WHERE 1=1" params = [] if query.project: sql += " AND project = $1" params.append(query.project) if query.node_type: sql += " AND type = $2" params.append(query.node_type) if query.name_contains: sql += " AND name LIKE $3" params.append(f"%{query.name_contains}%") async with app.state.pg.acquire() as conn: rows = await conn.fetch(sql, *params) nodes = [dict(row) for row in rows] # 缓存结果（5分钟过期） await app.state.redis.setex(cache_key, 300, json.dumps(nodes)) return {"nodes": nodes}@app.get("/project/{project_name}/dependencies")async def get_dependencies(project_name: str): """ 获取指定项目的依赖关系图。 """ async with app.state.pg.acquire() as conn: rows = await conn.fetch( "SELECT * FROM edges WHERE source IN (SELECT hash FROM nodes WHERE project = $1)", project_name ) edges = [dict(row) for row in rows] return {"edges": edges}@app.post("/ingest")async def ingest_graph(project: str, graph_data: dict): """ 接收 CodeGraph 引擎生成的图数据，写入数据库。 """ async with app.state.pg.acquire() as conn: # 批量插入节点 for node in graph_data.get('nodes', []): await conn.execute( "INSERT INTO nodes (hash, type, name, file, line, project) VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (hash) DO UPDATE SET file=$4, line=$5", node['hash'], node['type'], node['name'], node['file'], node['line'], project ) # 批量插入边 for edge in graph_data.get('edges', []): await conn.execute( "INSERT INTO edges (source, target, relation) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING", edge['source'], edge['target'], edge['relation'] ) # 清除相关缓存 await app.state.redis.delete(f"graph:{project}:*") return {"status": "ok", "project": project, "nodes_ingested": len(graph_data.get('nodes', []))}### 第四步：数据库初始化init.sql：sqlCREATE TABLE IF NOT EXISTS nodes ( hash VARCHAR(64) PRIMARY KEY, type VARCHAR(50), name TEXT, file TEXT, line INTEGER, project VARCHAR(100));CREATE TABLE IF NOT EXISTS edges ( id SERIAL PRIMARY KEY, source VARCHAR(64) REFERENCES nodes(hash), target VARCHAR(64) REFERENCES nodes(hash), relation VARCHAR(50));### 第五步：Docker Compose 编排docker-compose.yml：yamlversion: '3.8'services: db: image: postgres:14 environment: POSTGRES_USER: user POSTGRES_PASSWORD: password POSTGRES_DB: codegraph volumes: - ./init.sql:/docker-entrypoint-initdb.d/init.sql - pgdata:/var/lib/postgresql/data ports: - "5432:5432" redis: image: redis:7-alpine ports: - "6379:6379" codegraph-engine: build: ./codegraph-engine volumes: - ./repos:/repos:ro command: > sh -c "python engine.py /repos/project-a > /tmp/graph_a.json && python engine.py /repos/project-b > /tmp/graph_b.json && curl -X POST -H 'Content-Type: application/json' -d @/tmp/graph_a.json http://gateway:8000/ingest?project=project-a && curl -X POST -H 'Content-Type: application/json' -d @/tmp/graph_b.json http://gateway:8000/ingest?project=project-b" depends_on: - gateway gateway: build: ./mcp-gateway ports: - "8000:8000" environment: - DB_URL=postgresql://user:password@db:5432/codegraph - REDIS_URL=redis://redis:6379/0 depends_on: - db - redisvolumes: pgdata:### 第六步：构建与运行bash# 在根目录创建示例仓库mkdir -p repos/project-a repos/project-becho "def hello():\n pass" > repos/project-a/main.pyecho "from utils import helper\ndef run():\n helper()" > repos/project-b/app.py# 启动所有服务docker-compose up --build访问 http://localhost:8000/docs 查看 API 文档。## 测试验证使用 curl 验证网关：bash# 查询 project-a 的所有函数节点curl -X POST http://localhost:8000/query \ -H "Content-Type: application/json" \ -d '{"project": "project-a", "node_type": "function"}'# 获取 project-b 的依赖边curl http://localhost:8000/project/project-b/dependencies你会看到类似输出：json{"nodes": [{"hash": "abc...", "type": "function", "name": "hello", "file": "/repos/project-a/main.py", "line": 1}]}## 总结本文完整演示了如何使用 Docker 部署一个企业级的 CodeGraph 多项目统一 MCP 网关。通过将代码解析引擎与网关分离，结合 PostgreSQL 存储与 Redis 缓存，我们实现了高效的多项目代码图聚合与查询。核心要点包括：1. 模块化设计：CodeGraph 引擎负责解析，MCP 网关负责查询，职责清晰。2. 容器化部署：Docker Compose 一键编排，降低运维成本。3. 性能优化：Redis 缓存热点查询，避免频繁数据库访问。4. 可扩展性：支持新增项目只需挂载新仓库并调用 ingest 接口。该方案可直接用于企业的代码智能分析、AI 辅助开发、跨项目依赖追踪等场景。完整源码已附于文中，你可根据自身需求调整解析规则或扩展 API。未来可进一步集成消息队列（如 RabbitMQ）实现异步扫描，或接入权限控制满足安全合规要求。