eric/PMOCrazycoder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance sandbox functionality and MCP tools

Browse Source 
This commit enhances the sandbox functionality by:
- Adding a new jj_log tool to retrieve commit history
- Refactoring sandbox tools into dedicated modules
- Improving project management with session-based active projects
- Updating Docker configuration to properly mount the sandbox volume
- Adding proper error handling and documentation for all tools
- Updating dependencies and configuration files
This commit is contained in:
Eric Coissac
2026-01-25 21:26:22 +01:00
parent 766b90a548
commit bc66c794a5
13 changed files with 281 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/PMOCrazyCoder.py
View File

@@ -1,5 +1,5 @@
from crazy_mcp import mcp
from sandbox.mcp import install_project_tool
from crazy_mcp import Context, mcp
from crazy_mcp.tools import *
# Déclaration d'un outil
@@ -20,6 +20,13 @@ def additionner(a: float, b: float) -> float:
return a + b
@mcp.tool
async def increment(ctx: Context) -> tuple[int, str]:
count = ctx.get_state("count") or 0
ctx.set_state("count", count + 1)
return count + 1, ctx.session_id
# Lancement du serveur MCP
if __name__ == "__main__":
mcp.run(

2
src/crazy_mcp/__init__.py
View File

@@ -1,3 +1,3 @@
from fastmcp import FastMCP
from fastmcp import Context, FastMCP
mcp = FastMCP("PMO Crazy coder")

2
src/crazy_mcp/tools/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .jj import *
from .sandbox import *

33
src/crazy_mcp/tools/jj.py Normal file
View File

@@ -0,0 +1,33 @@
from crazy_mcp import Context, mcp
from jujutsu import jj_log
from sandbox import get_active_project, get_project_path
@mcp.tool(description="Get the commit history for a project")
def jj_log_tool(project_name: str | None = None, ctx: Context = None) -> dict:
"""
Get the commit history for a project.
Args:
project_name (str, optional): The name of the project. If None, uses the active project.
ctx (Context): The MCP context object containing session information
Returns:
dict: The commit history as JSON structure
"""
# If no project name is provided, try to get the active project
if project_name is None:
if ctx is None:
raise ValueError("ctx is required when project_name is None")
project_info = get_active_project(ctx.session_id)
if project_info is None:
raise ValueError("No active project set and no project name provided")
project_path = project_info[1]
else:
project_path = get_project_path(project_name)
if project_path is None:
raise ValueError(f"Project '{project_name}' not found")
# Call the jj_log function from jujutsu module
return jj_log(project_path)

89
src/crazy_mcp/tools/sandbox.py Normal file
View File

@@ -0,0 +1,89 @@
from crazy_mcp import Context, mcp
from sandbox import (
get_active_project,
install_project,
remove_project,
sandboxed_project_list,
set_active_project,
)
@mcp.tool(
description="Clone a Git repository into the sandbox. Note: Access to the cloned project is only possible through MCP tools of this server. WARNING: This operation creates a sandboxed environment that prevents direct filesystem access to the project contents. Any attempt to access project files outside of the MCP tools will be blocked by security restrictions."
)
def install_project_tool(url: str, project_name: str = None) -> str:
"""
Install a Git project into the sandbox environment.
Args:
url: The Git repository URL to clone
project_name: Optional project name (if None, derived from URL)
Returns:
str: Confirmation message with project name
"""
project_name = install_project(url, project_name)
return f"Project '{project_name}' installed successfully in sandbox"
@mcp.tool(
description="Remove a project from the sandbox. Note: Access to the cloned project is only possible through MCP tools of this server. WARNING: This operation removes the entire project sandboxed environment and all associated security restrictions will be enforced."
)
def remove_project_tool(project_name: str) -> str:
"""
Remove a project from the sandbox environment.
Args:
project_name (str): The name of the project to remove
Returns:
str: Confirmation message with project name
"""
success = remove_project(project_name)
if success:
return f"Project '{project_name}' removed successfully from sandbox"
else:
return f"Failed to remove project '{project_name}' from sandbox"
@mcp.tool(
description="List all projects available in the sandbox. Note: Direct filesystem access to project contents is forbidden. All project interactions must be done through MCP tools."
)
def list_sandboxed_projects() -> list:
"""
Get a list of all projects available in the sandbox.
Returns:
list: A list of project names
"""
return sandboxed_project_list()
@mcp.tool(description="Set the active project for a given session ID")
def set_active_project_tool(project_name: str, ctx: Context) -> bool:
"""
Set the active project for a given session ID.
Args:
project_name (str): The name of the project to set as active
ctx (Context): The MCP context object containing session information
Returns:
bool: True if successful, False if the project doesn't exist
"""
return set_active_project(ctx.session_id, project_name)
@mcp.tool(description="Get the active project for a given session ID")
def get_active_project_tool(ctx: Context) -> tuple[str, str] | None:
"""
Get the active project for a given session ID.
Args:
ctx (Context): The MCP context object containing session information
Returns:
tuple: (project_name, project_path) if session exists, None otherwise
"""
project_info = get_active_project(ctx.session_id)
return project_info

7
src/json_extract/__init__.py
View File

@@ -1,5 +1,6 @@
import json
import re
from ollama import prompt
@@ -21,8 +22,8 @@ def structure_to_json(text, schema):
full_prompt = f"""Transform the following structured text into JSON according to the given schema.
**Return only valid JSON, without any Markdown formatting :**
- no backticks,
- no code blocks,
- no backticks,
- no code blocks,
- no extra text
Input text:
@@ -32,7 +33,7 @@ Schema:
{schema_json}"""
# Send to Ollama
response = prompt("ollama:qwen3:0.6b", full_prompt, allow_pull=True)
response = prompt("qwen3:0.6b", full_prompt, allow_pull=True)
# Remove any markdown formatting (backticks, code blocks) that Ollama might include
# Only remove backticks at the beginning or end of the response, not those within the JSON

38
src/jujutsu/__init__.py
View File

@@ -1 +1,39 @@
import os
from sys import implementation
from json_extract import structure_to_json
from jujutsu.command import jj_log as jj_log_command
from .command import jj
def jj_log(working_dir=None):
"""
Execute jj log command for a project and return structured data as JSON.
Args:
project_path (str): The path of the project
Returns:
dict: JSON structure with commit history
"""
# Execute jj log command with working_dir parameter
log_output = jj_log_command(working_dir=working_dir)
# Define the schema for the JSON structure
schema = {
"type": "array",
"items": {
"type": "object",
"properties": {
"hash": {"type": "string"},
"author": {"type": "string"},
"date": {"type": "string"},
"message": {"type": "string"},
},
"required": ["hash", "author", "date", "message"],
},
}
# Convert the text output to JSON using the schema
return structure_to_json(log_output, schema)

46
src/jujutsu/command.py
View File

@@ -1,4 +1,5 @@
import json
import os
import subprocess
from typing import List, Optional
@@ -8,6 +9,7 @@ def jj(
options: Optional[List[str]] = None,
args: Optional[List[str]] = None,
template: Optional[str] = None,
working_dir: Optional[str] = None,
) -> str:
"""
Lance une commande jj.
@@ -17,6 +19,7 @@ def jj(
options: Liste des options (ex: ['--limit', '5'])
args: Liste des arguments additionnels
template: Template jj à utiliser (ex: 'builtin_log_compact')
working_dir: Répertoire de travail optionnel
Returns:
str: sortie de la commande
@@ -32,7 +35,16 @@ def jj(
if args:
cmd.extend(args)
result = subprocess.run(cmd, capture_output=True, text=True)
# Si un répertoire de travail est spécifié, on le change temporairement
if working_dir is not None:
original_cwd = os.getcwd()
try:
os.chdir(working_dir)
result = subprocess.run(cmd, capture_output=True, text=True)
finally:
os.chdir(original_cwd)
else:
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(
@@ -42,7 +54,7 @@ def jj(
return result.stdout
def jj_log(options=None, args=None, template=None):
def jj_log(options=None, args=None, template=None, working_dir: Optional[str] = None):
"""
Execute jj log command with optional arguments.
@@ -50,14 +62,15 @@ def jj_log(options=None, args=None, template=None):
options: List of options to pass to jj log (ex: ['--limit', '5'])
args: List of additional arguments
template: Template to use (ex: 'builtin_log_compact')
working_dir: Working directory for the command (optional)
Returns:
str: Output of the jj log command
"""
return jj("log", options, args, template)
return jj("log", options, args, template, working_dir=working_dir)
def jj_clone(url, destination=None, options=None):
def jj_clone(url, destination=None, options=None, working_dir: Optional[str] = None):
"""
Execute jj git clone command.
@@ -65,23 +78,26 @@ def jj_clone(url, destination=None, options=None):
url: The repository URL to clone
destination: The destination directory (optional)
options: List of additional options for jj git clone
working_dir: Working directory for the command (optional)
Returns:
str: Output of the jj git clone command
"""
cmd = ["jj", "git", "clone", url]
# For jj git clone, we build the full command and pass it to jj function
# The subcommand "git clone" is handled by jj itself
cmd_args = ["git", "clone", url]
if destination:
cmd.append(destination)
cmd_args.append(destination)
if options:
cmd.extend(options)
cmd_args.extend(options)
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(
f"Commande jj git clone échouée: {' '.join(cmd)}\nstderr: {result.stderr}"
)
return result.stdout
# Call the main jj function with subcommand "git clone"
return jj(
"git",
options=["clone", url],
args=destination,
template=None,
working_dir=working_dir,
)

69
src/sandbox/__init__.py
View File

@@ -1,7 +1,8 @@
import os
import re
from pathlib import Path
import shutil
import subprocess
from pathlib import Path
from jujutsu.command import jj_clone
@@ -10,6 +11,9 @@ __sandbox_root__ = "/sandbox"
# Create the sandbox root directory if it doesn't exist
os.makedirs(__sandbox_root__, exist_ok=True)
# Dictionary to store session information (session_id -> (project_name, project_path))
__session_projects__ = {}
def get_sandbox_root():
"""Return the sandbox root directory path."""
@@ -32,11 +36,8 @@ def install_project(url, project_name=None):
# Extract project name from URL (e.g., from 'https://github.com/user/repo.git' get 'repo')
project_name = re.sub(r".*/([^/]+?)(?:\.git)?$", r"\1", url)
# Create the full path for the project
project_path = os.path.join(__sandbox_root__, project_name)
# Clone the repository
jj_clone(url, destination=project_path)
# Clone the repository using the working_dir parameter to ensure it's cloned in the sandbox
jj_clone(url, working_dir=__sandbox_root__)
return project_name
@@ -80,3 +81,59 @@ def sandboxed_project_list():
pass
return projects
def set_active_project(session_id, project_name):
"""
Set the active project for a given session ID.
Args:
session_id (str): The session identifier
project_name (str): The name of the project to set as active
Returns:
bool: True if successful, False if the project doesn't exist
"""
# Check if the project exists
projects = sandboxed_project_list()
if project_name not in projects:
return False
# Get the project path
project_path = os.path.join(__sandbox_root__, project_name)
# Store the project information in the session dictionary
__session_projects__[session_id] = (project_name, project_path)
return True
def get_project_path(project_name):
"""
Get the path of a project on the disk.
Args:
project_name (str): The name of the project
Returns:
str: The full path to the project directory
"""
return os.path.join(__sandbox_root__, project_name)
def get_active_project(session_id):
"""
Get the active project for a given session ID.
Args:
session_id (str): The session identifier
Returns:
tuple: (project_name, project_path) if session exists, None otherwise
"""
# Check if the session ID exists
if session_id not in __session_projects__:
return None
# Return the project information
return __session_projects__[session_id]

51
src/sandbox/mcp.py
View File

@@ -1,51 +0,0 @@
from sandbox import install_project, sandboxed_project_list, remove_project
from crazy_mcp import mcp
@mcp.tool(
description="Clone a Git repository into the sandbox. Note: Access to the cloned project is only possible through MCP tools of this server."
)
def install_project_tool(url: str, project_name: str = None) -> str:
"""
Install a Git project into the sandbox environment.
Args:
url: The Git repository URL to clone
project_name: Optional project name (if None, derived from URL)
Returns:
str: Confirmation message with project name
"""
project_name = install_project(url, project_name)
return f"Project '{project_name}' installed successfully in sandbox"
@mcp.tool(
description="Remove a project from the sandbox. Note: Access to the cloned project is only possible through MCP tools of this server."
)
def remove_project_tool(project_name: str) -> str:
"""
Remove a project from the sandbox environment.
Args:
project_name (str): The name of the project to remove
Returns:
str: Confirmation message with project name
"""
success = remove_project(project_name)
if success:
return f"Project '{project_name}' removed successfully from sandbox"
else:
return f"Failed to remove project '{project_name}' from sandbox"
@mcp.tool(description="List all projects available in the sandbox")
def list_sandboxed_projects() -> list:
"""
Get a list of all projects available in the sandbox.
Returns:
list: A list of project names
"""
return sandboxed_project_list()