enhance: move file to correct dir
This commit is contained in:
parent
adf31970ab
commit
d3ac93a0df
148
ollama/mcp.py
Normal file
148
ollama/mcp.py
Normal file
@ -0,0 +1,148 @@
|
||||
# /// script
|
||||
# dependencies = [
|
||||
# "mcp",
|
||||
# "google-generativeai"
|
||||
# ]
|
||||
# ///
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
import google.generativeai as genai
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
import random
|
||||
|
||||
ALLOWED_ROOT = "/app"
|
||||
|
||||
API_KEY = os.environ.get("GEMINI_API_KEY") or "TOKEN"
|
||||
genai.configure(api_key=API_KEY)
|
||||
|
||||
mcp = FastMCP("Gemini Go Backend")
|
||||
|
||||
@mcp.tool()
|
||||
def ask_gemini(prompt: str, model_name: str = "gemini-3-pro") -> str:
|
||||
"""
|
||||
向 Google Gemini 提問。建議優先使用 'gemini-3-pro' 以獲得更快的速度與更高的額度限制。
|
||||
"""
|
||||
try:
|
||||
model = genai.GenerativeModel(model_name)
|
||||
response = model.generate_content(prompt)
|
||||
return response.text
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
if "429" in error_msg:
|
||||
return f"Error: API 額度限制 (429)。請嘗試切換為 'gemini-2.0-flash-exp' 或稍後再試。"
|
||||
return f"Error calling Gemini: {error_msg}"
|
||||
|
||||
@mcp.tool()
|
||||
def read_file(filepath: str) -> str:
|
||||
"""讀取專案內的檔案內容。"""
|
||||
full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filepath))
|
||||
if not full_path.startswith(ALLOWED_ROOT):
|
||||
return "Error: Access denied. Path outside allowed root."
|
||||
|
||||
try:
|
||||
with open(full_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
except Exception as e:
|
||||
return f"Error reading file: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
def write_file(filepath: str, content: str) -> str:
|
||||
"""覆蓋寫入檔案。檔案不存在時會自動建立。"""
|
||||
full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filepath))
|
||||
os.makedirs(os.path.dirname(full_path), exist_ok=True)
|
||||
|
||||
try:
|
||||
with open(full_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return f"Success: File written to {filepath}"
|
||||
except Exception as e:
|
||||
return f"Error writing file: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
def append_file(filepath: str, content: str) -> str:
|
||||
"""分段追加內容到檔案末尾。適用於解決輸出過長的問題。"""
|
||||
full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filepath))
|
||||
mode = "a" if os.path.exists(full_path) else "w"
|
||||
|
||||
try:
|
||||
with open(full_path, mode, encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return f"Success: Content appended to {filepath}"
|
||||
except Exception as e:
|
||||
return f"Error appending file: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
def list_files(directory: str = ".") -> str:
|
||||
"""列出目錄下的所有檔案。"""
|
||||
full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, directory))
|
||||
try:
|
||||
files = os.listdir(full_path)
|
||||
return "\n".join(files)
|
||||
except Exception as e:
|
||||
return f"Error listing directory: {str(e)}"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def append_file(filepath: str, content: str) -> str:
|
||||
"""
|
||||
將內容「追加」到檔案末尾。
|
||||
自動在內容末尾補上換行符號,確保分段寫入的程式碼結構正確。
|
||||
"""
|
||||
full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filepath))
|
||||
|
||||
if not content.endswith('\n'):
|
||||
content += '\n'
|
||||
|
||||
try:
|
||||
with open(full_path, "a", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return f"Success: Content appended to {filepath}"
|
||||
except Exception as e:
|
||||
return f"Error appending file: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
def backup_file(filepath: str) -> str:
|
||||
"""
|
||||
為指定檔案建立帶有時間戳記的備份。
|
||||
例如: main.go -> main.go.20231217_103000.bak
|
||||
"""
|
||||
src_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filepath))
|
||||
if not os.path.exists(src_path):
|
||||
return f"Error: Source file {filepath} not found"
|
||||
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_filename = f"{filepath}.{timestamp}.bak"
|
||||
dest_path = os.path.join(ALLOWED_ROOT, backup_filename)
|
||||
|
||||
try:
|
||||
shutil.copy2(src_path, dest_path)
|
||||
return f"Success: Backup created at {backup_filename}"
|
||||
except Exception as e:
|
||||
return f"Error creating backup: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
def run_command(command: str) -> str:
|
||||
"""
|
||||
在專案目錄執行指令 (如: 'go test ./...', 'go mod tidy')。
|
||||
僅允許安全的白名單指令。
|
||||
"""
|
||||
safe_commands = ["go ", "ls ", "cat ", "mkdir ", "rm ", "cp ", "mv ", "echo ", "pwd ", "git ","head ","tail ", "wc "]
|
||||
if not any(command.startswith(cmd) for cmd in safe_commands):
|
||||
return "Error: 指令不在安全白名單內。"
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command, shell=True, cwd=ALLOWED_ROOT,
|
||||
capture_output=True, text=True, timeout=60
|
||||
)
|
||||
output = f"Stdout:\n{result.stdout}"
|
||||
if result.stderr:
|
||||
output += f"\nStderr:\n{result.stderr}"
|
||||
return output
|
||||
except Exception as e:
|
||||
return f"Error executing command: {str(e)}"
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run()
|
||||
Loading…
x
Reference in New Issue
Block a user