Add templates and MCP server modules
This commit is contained in:
133
src/contextgen/mcp/server.py
Normal file
133
src/contextgen/mcp/server.py
Normal file
@@ -0,0 +1,133 @@
|
|||||||
|
"""MCP server implementation for AI tool integration."""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from contextgen.context_generator import ContextGenerator
|
||||||
|
|
||||||
|
|
||||||
|
class MCPServer:
|
||||||
|
"""Model Context Protocol server for AI tool integration."""
|
||||||
|
|
||||||
|
def __init__(self, project_path: Path):
|
||||||
|
self.project_path = Path(project_path).resolve()
|
||||||
|
self.context_generator = ContextGenerator(self.project_path)
|
||||||
|
self.context = self.context_generator.generate()
|
||||||
|
|
||||||
|
def get_project_overview(self) -> dict[str, Any]:
|
||||||
|
"""Get project overview information."""
|
||||||
|
return {
|
||||||
|
"name": self.context["project"]["name"],
|
||||||
|
"description": self.context["project"]["description"],
|
||||||
|
"primary_language": self.context["analysis"].get("primary_language"),
|
||||||
|
"total_files": self.context["structure"]["total_files"],
|
||||||
|
"frameworks": [
|
||||||
|
fw["name"] for fw in self.context["analysis"].get("frameworks", [])
|
||||||
|
],
|
||||||
|
"build_tools": [
|
||||||
|
tool["name"] for tool in self.context["analysis"].get("build_tools", [])
|
||||||
|
],
|
||||||
|
}
|
||||||
|
|
||||||
|
def get_file_context(self, file_path: str) -> dict[str, Any]:
|
||||||
|
"""Get context for a specific file."""
|
||||||
|
full_path = self.project_path / file_path
|
||||||
|
|
||||||
|
if not full_path.exists():
|
||||||
|
return {"error": f"File not found: {file_path}"}
|
||||||
|
|
||||||
|
if full_path.is_file():
|
||||||
|
content = full_path.read_text(encoding="utf-8")
|
||||||
|
return {
|
||||||
|
"path": str(full_path.relative_to(self.project_path)),
|
||||||
|
"exists": True,
|
||||||
|
"is_file": True,
|
||||||
|
"size": full_path.stat().st_size,
|
||||||
|
"content_preview": content[:1000],
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
return {
|
||||||
|
"path": str(full_path.relative_to(self.project_path)),
|
||||||
|
"exists": True,
|
||||||
|
"is_file": False,
|
||||||
|
}
|
||||||
|
|
||||||
|
def get_conventions(self) -> dict[str, Any]:
|
||||||
|
"""Get coding conventions for the project."""
|
||||||
|
return self.context.get("conventions", {})
|
||||||
|
|
||||||
|
def search_structure(self, query: str) -> dict[str, Any]:
|
||||||
|
"""Search project structure for matching files."""
|
||||||
|
results: list[dict[str, str]] = []
|
||||||
|
|
||||||
|
key_files = self.context["structure"].get("key_files", {})
|
||||||
|
for filename, description in key_files.items():
|
||||||
|
if query.lower() in filename.lower() or query.lower() in description.lower():
|
||||||
|
results.append({
|
||||||
|
"file": filename,
|
||||||
|
"description": description,
|
||||||
|
})
|
||||||
|
|
||||||
|
return {"query": query, "results": results}
|
||||||
|
|
||||||
|
def get_full_context(self) -> dict[str, Any]:
|
||||||
|
"""Get the complete project context."""
|
||||||
|
return self.context
|
||||||
|
|
||||||
|
def run_stdio(self) -> None:
|
||||||
|
"""Run the MCP server using stdio transport."""
|
||||||
|
import argparse
|
||||||
|
|
||||||
|
parser = argparse.ArgumentParser(description="Project Context Generator MCP Server")
|
||||||
|
parser.add_argument("--project", type=str, required=True, help="Path to project directory")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
self.project_path = Path(args.project).resolve()
|
||||||
|
self.context_generator = ContextGenerator(self.project_path)
|
||||||
|
self.context = self.context_generator.generate()
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
line = sys.stdin.readline()
|
||||||
|
if not line:
|
||||||
|
break
|
||||||
|
|
||||||
|
request = json.loads(line.strip())
|
||||||
|
response = self._handle_request(request)
|
||||||
|
print(json.dumps(response))
|
||||||
|
sys.stdout.flush()
|
||||||
|
except (json.JSONDecodeError, KeyboardInterrupt):
|
||||||
|
break
|
||||||
|
|
||||||
|
def _handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
|
||||||
|
"""Handle an MCP request."""
|
||||||
|
method = request.get("method")
|
||||||
|
params = request.get("params", {})
|
||||||
|
|
||||||
|
if method == "get_project_overview":
|
||||||
|
return {"result": self.get_project_overview()}
|
||||||
|
elif method == "get_file_context":
|
||||||
|
return {"result": self.get_file_context(params.get("path", ""))}
|
||||||
|
elif method == "get_conventions":
|
||||||
|
return {"result": self.get_conventions()}
|
||||||
|
elif method == "search_structure":
|
||||||
|
return {"result": self.search_structure(params.get("query", ""))}
|
||||||
|
elif method == "get_full_context":
|
||||||
|
return {"result": self.get_full_context()}
|
||||||
|
else:
|
||||||
|
return {"error": f"Unknown method: {method}"}
|
||||||
|
|
||||||
|
|
||||||
|
def create_mcp_handlers(project_path: Path):
|
||||||
|
"""Create MCP handler functions for a project."""
|
||||||
|
server = MCPServer(project_path)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"get_project_overview": server.get_project_overview,
|
||||||
|
"get_file_context": server.get_file_context,
|
||||||
|
"get_conventions": server.get_conventions,
|
||||||
|
"search_structure": server.search_structure,
|
||||||
|
"get_full_context": server.get_full_context,
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user