104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
mcp_builder.py — Générateur de serveurs MCP auto-détectés
|
|
Scan un dossier projet → génère un serveur MCP stdio/SSE
|
|
"""
|
|
import os, sys, json, ast, argparse
|
|
from pathlib import Path
|
|
from typing import List, Dict
|
|
|
|
TEMPLATES = {
|
|
"python": """#!/usr/bin/env python3
|
|
import asyncio, json
|
|
from mcp.server import Server
|
|
from mcp.types import TextContent
|
|
|
|
server = Server("{name}")
|
|
|
|
{tools}
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(server.run_stdio_async())
|
|
""",
|
|
"nodejs": """#!/usr/bin/env node
|
|
const { Server } = require('@modelcontextprotocol/sdk/server/index.js');
|
|
const { StdioServerTransport } = require('@modelcontextprotocol/sdk/server/stdio.js');
|
|
|
|
const server = new Server({{ name: "{name}" }}, {{
|
|
capabilities: {{ tools: {} }}
|
|
}});
|
|
|
|
{tools}
|
|
|
|
async function main() {{
|
|
const transport = new StdioServerTransport();
|
|
await server.connect(transport);
|
|
}}
|
|
main();
|
|
"""
|
|
}
|
|
|
|
def scan_functions(path: str) -> List[Dict]:
|
|
"""Scan les fichiers Python/JS pour détecter fonctions publiques."""
|
|
tools = []
|
|
root = Path(path)
|
|
for f in root.rglob("*.py"):
|
|
try:
|
|
tree = ast.parse(f.read_text())
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef) and not node.name.startswith("_"):
|
|
args = [a.arg for a in node.args.args]
|
|
tools.append({
|
|
"name": node.name,
|
|
"file": str(f.relative_to(root)),
|
|
"args": args,
|
|
"language": "python"
|
|
})
|
|
except:
|
|
pass
|
|
return tools
|
|
|
|
def generate_mcp(tools: List[Dict], name: str, lang: str = "python") -> str:
|
|
"""Génère le code serveur MCP."""
|
|
tool_blocks = []
|
|
for t in tools[:10]: # limite pour éviter overload
|
|
if lang == "python":
|
|
args_schema = ", ".join([f'"{a}": {{"type": "string"}}' for a in t["args"]])
|
|
tool_blocks.append(f'''
|
|
@server.tool("{t["name"]}")
|
|
async def {t["name"]}({", ".join(t["args"])}):
|
|
"""Auto-généré depuis {t["file"]}"""
|
|
result = "TODO: implémenter"
|
|
return [TextContent(type="text", text=result)]
|
|
''')
|
|
else:
|
|
tool_blocks.append(f'''
|
|
server.setRequestHandler("tools/call", async (request) => {{
|
|
if (request.params.name === "{t["name"]}") {{
|
|
return {{ content: [{{ type: "text", text: "TODO" }}] }};
|
|
}}
|
|
}});
|
|
''')
|
|
|
|
template = TEMPLATES.get(lang, TEMPLATES["python"])
|
|
return template.format(name=name, tools="\n".join(tool_blocks))
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(prog="mcp-builder")
|
|
parser.add_argument("scan", nargs="?", default=".", help="Dossier à scanner")
|
|
parser.add_argument("--name", default="shango-mcp", help="Nom du serveur MCP")
|
|
parser.add_argument("--lang", choices=["python", "nodejs"], default="python")
|
|
parser.add_argument("--out", default="mcp_server.py", help="Fichier de sortie")
|
|
args = parser.parse_args()
|
|
|
|
print(f"[MCP-BUILDER] Scan de {args.scan}...")
|
|
tools = scan_functions(args.scan)
|
|
print(f"[MCP-BUILDER] {len(tools)} fonctions détectées")
|
|
|
|
code = generate_mcp(tools, args.name, args.lang)
|
|
Path(args.out).write_text(code)
|
|
print(f"[MCP-BUILDER] Serveur généré: {args.out} ({len(tools)} tools)")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|