api.py•10.5 kB
"""
API路由模块
定义MCP代码检索服务的API路由
"""
import os
import logging
from typing import Dict, Any, List, Optional
from flask import Flask, jsonify, request, current_app
logger = logging.getLogger(__name__)
def setup_routes(app: Flask) -> None:
"""
设置API路由
Args:
app: Flask应用实例
Returns:
无返回值
"""
# 健康检查
@app.route('/health', methods=['GET'])
def health_check():
"""
健康检查接口
Returns:
健康状态响应
"""
return jsonify({
'status': 'ok',
'version': '0.1.0'
})
# 项目识别
@app.route('/api/project/identify', methods=['POST'])
def identify_project():
"""
项目识别接口
请求体:
{
"project_path": "项目路径"
}
Returns:
项目识别结果
"""
data = request.json
if not data or 'project_path' not in data:
return jsonify({
'error': '无效请求',
'message': '缺少项目路径'
}), 400
project_path = data['project_path']
if not os.path.exists(project_path):
return jsonify({
'error': '无效的项目路径',
'message': f'路径不存在：{project_path}'
}), 400
try:
# 获取组件
indexer = current_app.config['mcp_indexer']
formatter = current_app.config['mcp_formatter']
# 识别项目
project_id, is_new, metadata = indexer.project_identifier.identify_project(project_path)
# 获取索引状态
status, progress = indexer.get_indexing_status(project_id)
# 格式化响应
response = formatter.format_project_info({
'project_id': project_id,
'project_path': project_path,
'is_new': is_new,
'status': status,
'progress': progress,
'metadata': metadata
})
return jsonify(response)
except Exception as e:
logger.error(f"项目识别失败: {str(e)}")
return jsonify({
'error': '项目识别失败',
'message': str(e)
}), 500
# 索引项目
@app.route('/api/project/index', methods=['POST'])
def index_project():
"""
索引项目接口
请求体:
{
"project_path": "项目路径"
}
Returns:
索引启动结果
"""
data = request.json
if not data or 'project_path' not in data:
return jsonify({
'error': 'Invalid request',
'message': 'Missing project_path'
}), 400
project_path = data['project_path']
if not os.path.exists(project_path):
return jsonify({
'error': 'Invalid project path',
'message': f'Path does not exist: {project_path}'
}), 400
try:
# 获取组件
indexer = current_app.config['mcp_indexer']
formatter = current_app.config['mcp_formatter']
# 启动索引
project_id = indexer.index_project(project_path)
# 获取索引状态
status, progress = indexer.get_indexing_status(project_id)
# 格式化响应
response = formatter.format_indexing_status(
project_id, status, progress,
"索引启动成功"
)
return jsonify(response)
except Exception as e:
logger.error(f"启动索引失败: {str(e)}")
return jsonify({
'error': '索引失败',
'message': str(e)
}), 500
# 获取索引状态
@app.route('/api/project/status/<project_id>', methods=['GET'])
def get_project_status(project_id):
"""
获取项目索引状态接口
Args:
project_id: 项目ID
Returns:
项目索引状态
"""
try:
# 获取组件
indexer = current_app.config['mcp_indexer']
formatter = current_app.config['mcp_formatter']
# 获取索引状态
status, progress = indexer.get_indexing_status(project_id)
# 格式化响应
response = formatter.format_indexing_status(
project_id, status, progress
)
return jsonify(response)
except Exception as e:
logger.error(f"获取索引状态失败: {str(e)}")
return jsonify({
'error': '获取状态失败',
'message': str(e)
}), 500
# 搜索代码
@app.route('/api/search', methods=['POST'])
def search_code():
"""
搜索代码接口
请求体:
{
"query": "查询字符串",
"project_ids": ["项目ID1", "项目ID2"], // 可选
"filters": { // 可选
"language": "python",
"file_path": "src"
},
"limit": 10 // 可选
}
Returns:
搜索结果
"""
data = request.json
if not data or 'query' not in data:
return jsonify({
'error': '无效请求',
'message': '缺少查询参数'
}), 400
query = data['query']
project_ids = data.get('project_ids')
filters = data.get('filters')
limit = data.get('limit', 10)
try:
# 获取组件
search_engine = current_app.config['mcp_search_engine']
formatter = current_app.config['mcp_formatter']
# 执行搜索
results = search_engine.search(query, project_ids, filters, limit)
# 格式化响应
response = formatter.format_search_results(results, query)
return jsonify(response)
except Exception as e:
logger.error(f"搜索代码失败: {str(e)}")
return jsonify({
'error': '搜索失败',
'message': str(e)
}), 500
# 获取代码上下文
@app.route('/api/context', methods=['POST'])
def get_code_context():
"""
获取代码上下文接口
请求体:
{
"file_path": "文件路径",
"line_number": 10,
"context_lines": 5 // 可选
}
Returns:
代码上下文
"""
data = request.json
if not data or 'file_path' not in data or 'line_number' not in data:
return jsonify({
'error': '无效请求',
'message': '缺少文件路径或行号'
}), 400
file_path = data['file_path']
line_number = data['line_number']
context_lines = data.get('context_lines', 10)
if not os.path.exists(file_path):
return jsonify({
'error': '无效的文件路径',
'message': f'文件不存在：{file_path}'
}), 400
try:
# 获取组件
search_engine = current_app.config['mcp_search_engine']
formatter = current_app.config['mcp_formatter']
# 获取代码上下文
context = search_engine.get_code_context(file_path, line_number, context_lines)
# 获取相关代码
related_blocks = search_engine.get_related_code(context, 5)
# 格式化响应
response = formatter.format_code_context(context, related_blocks)
return jsonify(response)
except Exception as e:
logger.error(f"获取代码上下文失败: {str(e)}")
return jsonify({
'error': '获取代码上下文失败',
'message': str(e)
}), 500
# 获取项目列表
@app.route('/api/projects', methods=['GET'])
def get_projects():
"""
获取项目列表接口
Returns:
项目列表
"""
try:
# 获取组件
indexer = current_app.config['mcp_indexer']
# 获取项目列表
projects = indexer.get_indexed_projects()
return jsonify({
'projects': projects,
'count': len(projects)
})
except Exception as e:
logger.error(f"获取项目列表失败: {str(e)}")
return jsonify({
'error': '获取项目列表失败',
'message': str(e)
}), 500
# 删除项目索引
@app.route('/api/project/<project_id>', methods=['DELETE'])
def delete_project(project_id):
"""
删除项目索引接口
Args:
project_id: 项目ID
Returns:
删除结果
"""
try:
# 获取组件
indexer = current_app.config['mcp_indexer']
# 删除项目索引
success = indexer.delete_project_index(project_id)
if success:
return jsonify({
'success': True,
'message': f'项目 {project_id} 删除成功'
})
else:
return jsonify({
'success': False,
'message': f'删除项目 {project_id} 失败'
}), 400
except Exception as e:
logger.error(f"删除项目索引失败: {str(e)}")
return jsonify({
'error': '删除项目失败',
'message': str(e)
}), 500
logger.info("API路由设置完成")