Chris Christiansen
9fcb9c354a
Some checks are pending
Check Python Version Consistency / Check Python Version (push) Waiting to run
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
import base64
|
|
import os
|
|
import requests
|
|
import ast
|
|
from google.cloud import secretmanager
|
|
|
|
def get_gitea_api_token():
|
|
"""Fetches the Gitea API token from GCP Secret Manager."""
|
|
try:
|
|
client = secretmanager.SecretManagerServiceClient()
|
|
secret_name = "gitea-api-token"
|
|
project_id = os.environ.get("GCP_PROJECT")
|
|
resource_name = f"projects/{project_id}/secrets/{secret_name}/versions/latest"
|
|
response = client.access_secret_version(name=resource_name)
|
|
return response.payload.data.decode("UTF-8")
|
|
except Exception as e:
|
|
print(f"Error fetching Gitea API token: {e}")
|
|
return None
|
|
|
|
def generate_tool_code(tool_name, tool_spec):
|
|
"""Generates boilerplate Python code for a new tool."""
|
|
|
|
params = tool_spec.get('parameters', {})
|
|
param_list = []
|
|
for param_name, param_type in params.items():
|
|
param_list.append(f"{param_name}: {param_type}")
|
|
|
|
param_str = ", ".join(param_list)
|
|
|
|
lines = [
|
|
"# This is an auto-generated tool file.",
|
|
"# It was created by the Gitea Provisioning Factory.",
|
|
"",
|
|
f"def {tool_name}({param_str}):",
|
|
f' """',
|
|
f' This is a new tool called {tool_name}.',
|
|
f' ',
|
|
f' Args:',
|
|
f' {", ".join(params.keys())}',
|
|
f' """',
|
|
f' ',
|
|
f' print(f"Executing tool {tool_name} with parameters: {{", ".join(params.keys())}})',
|
|
f' ',
|
|
f' # Your tool logic goes here.',
|
|
f' ',
|
|
f' return {{"status": "success", "message": "Tool {tool_name} executed successfully."}}',
|
|
"",
|
|
]
|
|
code = "\n".join(lines)
|
|
return code
|
|
|
|
def provision_new_mcp_module(module_name: str, tool_name: str, tool_spec: dict, client_id: str = None):
|
|
"""
|
|
Provisions a new MCP module by generating a tool file and committing it to Gitea.
|
|
"""
|
|
gitea_token = get_gitea_api_token()
|
|
if not gitea_token:
|
|
return {"status": "error", "message": "Failed to get Gitea API token."}
|
|
|
|
tool_code = generate_tool_code(tool_name, tool_spec)
|
|
|
|
# Pre-flight syntax check
|
|
try:
|
|
ast.parse(tool_code)
|
|
except SyntaxError as e:
|
|
return {"status": "error", "message": f"Generated code failed syntax check: {e}"}
|
|
|
|
gitea_api_url = "https://git.vauco.no/api/v1/repos/chris_christiansen/OSVauco/contents"
|
|
file_path = f"modules/{module_name}/tools/{tool_name}.py"
|
|
|
|
# Check if file exists
|
|
response = requests.get(f"{gitea_api_url}/{file_path}", headers={"Authorization": f"token {gitea_token}"})
|
|
|
|
sha = None
|
|
if response.status_code == 200:
|
|
sha = response.json().get('sha')
|
|
|
|
message = f"feat: Add new tool {tool_name} to module {module_name}"
|
|
data = {
|
|
"content": base64.b64encode(tool_code.encode()).decode(),
|
|
"message": message
|
|
}
|
|
if sha:
|
|
data['sha'] = sha
|
|
|
|
response = requests.put(
|
|
f"{gitea_api_url}/{file_path}",
|
|
headers={"Authorization": f"token {gitea_token}"},
|
|
json=data
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
redeploy_command = "gcloud run deploy osvx-mcp --source ./opax-mcp --region us-central1 --allow-unauthenticated '--set-env-vars \"MCP_SECRET=$(gcloud secrets versions access latest --secret=osvx-mcp-api-key)\"'"
|
|
print("Tool provisioned successfully. To redeploy, run the following command:")
|
|
print(redeploy_command)
|
|
return {"status": "success", "message": f"Tool {tool_name} provisioned successfully.", "redeploy_command": redeploy_command}
|
|
else:
|
|
return {"status": "error", "message": f"Failed to commit to Gitea: {response.text}"}
|
|
|
|
if __name__ == '__main__':
|
|
# Example usage:
|
|
spec = {
|
|
"parameters": {
|
|
"param1": "string",
|
|
"param2": "int"
|
|
}
|
|
}
|
|
provision_new_mcp_module("test_module", "my_new_tool", spec)
|