push-umtslxtpqxls #1

Merged
eric merged 5 commits from push-umtslxtpqxls into main 2026-01-31 10:36:47 +01:00
21 changed files with 846 additions and 2 deletions

8
.gitignore vendored
View File

@@ -1,3 +1,7 @@
reseau
context
bac_a_sable
# ---> macOS # ---> macOS
# General # General
.DS_Store .DS_Store
@@ -5,7 +9,8 @@
.LSOverride .LSOverride
# Icon must end with two \r # Icon must end with two \r
Icon Icon
# Thumbnails # Thumbnails
._* ._*
@@ -201,4 +206,3 @@ cython_debug/
# PyPI configuration file # PyPI configuration file
.pypirc .pypirc
.pypirc

12
.zed/settings.json Normal file
View File

@@ -0,0 +1,12 @@
// Folder-specific settings
//
// For a full list of overridable settings, and general information on folder-specific settings,
// see the documentation: https://zed.dev/docs/configuring-zed#settings-files
{
"inlay_hints": {
"enabled": true
},
"colorize_brackets": true,
"project_name": "PMO Crazy Coder",
"autosave": "on_window_change",
}

83
Dockerfile Normal file
View File

@@ -0,0 +1,83 @@
# Dockerfile
FROM python:3.11-slim
# Ajouter le dépôt Debian Testing pour les paquets plus récents
RUN echo "deb http://deb.debian.org/debian testing main" \
> /etc/apt/sources.list.d/testing.list
# Création d'un utilisateur programmeur avec UID 1000 et home dans /home/programmer
RUN useradd -u 1000 -m -d /home/programmer -s /bin/bash programmer
WORKDIR /app
# Installer les outils de développement Go et Rust
RUN apt-get update && \
apt-get install -y \
clang \
llvm-dev \
libclang-dev \
pkg-config \
build-essential \
git \
make \
curl \
nodejs \
npm \
pylint \
black \
flake8 \
mypy \
isort \
logrotate
COPY tools tools
# Installer la dernière version de Go
RUN bash tools/install_latest_go.sh
ENV PATH=/usr/local/go/bin:$PATH
# Installer la dernière version de Rust
ENV RUSTUP_HOME=/usr/local/rustup
ENV CARGO_HOME=/usr/local/cargo
ENV PATH=/usr/local/cargo/bin:$PATH
RUN curl https://sh.rustup.rs -sSf \
| sh -s -- -y --default-toolchain stable \
&& rustup install nightly \
&& rustup component add rustfmt clippy rust-src rust-docs \
&& cargo install cargo-edit cargo-watch cargo-outdated cargo-make cargo-binstall
# Installer la dernière version de typscript
RUN npm install -g typescript tsc @types/node eslint prettier
RUN cargo binstall --strategies crate-meta-data jj-cli
# Installer le SDK MCP Python
RUN pip install --no-cache-dir ruff fastmcp ipython
RUN curl -fsSL https://opencode.ai/install | bash \
&& mv /root/.opencode/bin/opencode /usr/local/bin/
# Copier le serveur
COPY src .
# Copier le script de démarrage
COPY docker_startup.sh .
COPY opencode.json ~/.config/opencode
# Exposer le port
EXPOSE 7860
ENV OLLAMA_ADDR="http://host.docker.internal:11434"
# Rendre le script exécutable et lancer le serveur
RUN chmod +x docker_startup.sh
RUN mkdir /sandbox /log \
&& chown programmer:programmer /sandbox /log
# Création de la configuration logrotate
COPY logrotate-config /etc/logrotate.d/mcp-logs
# Changer l'utilisateur pour programmeur
USER programmer
RUN mkdir -p /home/programmer/.config/opencode
COPY opencode.json /home/programmer/.config/opencode
CMD ["./docker_startup.sh"]

36
Makefile Normal file
View File

@@ -0,0 +1,36 @@
.PHONY: build run all debug
# Variables
IMAGE_NAME = pmo_crazy_coder
TAG = latest
PORT = 7860
SANDBOX_VOLUME = "./bac_a_sable"
CONTAINER_NAME = $(IMAGE_NAME)
DOCKER_RUN_COMMON = --rm -p $(PORT):$(PORT) -v $(SANDBOX_VOLUME):/sandbox --name $(CONTAINER_NAME)
FULL_IMAGE_NAME = $(IMAGE_NAME):$(TAG)
all: run
$(SANDBOX_VOLUME):
mkdir -p $(SANDBOX_VOLUME)
stampdir:
mkdir -p stamp
stamp/build_PMOCrazyCoder: stampdir Dockerfile
@echo building the PMOCrazyCoder server
@docker build -t $(FULL_IMAGE_NAME) .
@touch $@
build: stamp/build_PMOCrazyCoder
run: build $(SANDBOX_VOLUME)
@echo Launching the PMOCrazyCoder MCP server
@docker run $(DOCKER_RUN_COMMON) \
-it \
$(FULL_IMAGE_NAME)
debug: build $(SANDBOX_VOLUME)
@echo Launching the PMOCrazyCoder MCP server in debug mode
@docker run $(DOCKER_RUN_COMMON) \
-it \
$(FULL_IMAGE_NAME) /bin/bash

55
Notes/jj_to_json.md Normal file
View File

@@ -0,0 +1,55 @@
```bash
jj log | orla agent -m ollama:qwen3:0.6b \
"
Transform the following `jj log` output into JSON according to the given schema.
**Return only valid JSON, without any Markdown formatting :**
- no backticks,
- no code blocks,
- no extra text
Schema:
{
\"type\": \"array\",
\"items\": {
\"type\": \"object\",
\"properties\": {
\"hash\": {\"type\": \"string\"},
\"author\": {\"type\": \"string\"},
\"date\": {\"type\": \"string\"},
\"message\": {\"type\": \"string\"}
},
\"required\": [\"hash\", \"author\", \"date\", \"message\"]
}
}
" | grep -v '```' | jq .
```
```json
[
{
"hash": "252b5de3",
"author": "nrrwwpms",
"date": "2026-01-25 17:36:18",
"message": "(no description set)"
},
{
"hash": "9f9d9bc3",
"author": "stxkzlul",
"date": "2026-01-25 12:08:12",
"message": "Ajout de Dockerfile et configuration MCP"
},
{
"hash": "7412b51a",
"author": "xnrxukn",
"date": "2026-01-25 07:53:42",
"message": "Initial commit"
}
]
```
Ok, on va essayer de développer un outil un peu plus avancé.
Dans le fichier @src/sandbox/__init__.py Nous allons créer une fonction jj_log.
Cette fonction prendra en paramètres:
- un projet

13
docker_startup.sh Executable file
View File

@@ -0,0 +1,13 @@
#!/bin/bash
# Docker Startup Script for MCP Server and Open Code
# This script launches the MCP server in background and starts Open Code in interactive mode
echo "Starting MCP server in background..."
python3 PMOCrazyCoder.py >/log/server.log 2>&1 &
# Give the server a moment to start
sleep 3
echo "Starting Open Code in interactive mode..."
opencode

13
logrotate-config Normal file
View File

@@ -0,0 +1,13 @@
/log/server.log /log/opencode.log {
daily
rotate 5
compress
delaycompress
missingok
notifempty
create 644 programmer programmer
sharedscripts
postrotate
# No restart needed for log files
endscript
}

9
opencode.json Normal file
View File

@@ -0,0 +1,9 @@
{
"$schema": "https://opencode.ai/config.json",
"mcp": {
"PMO-Crazy-coder": {
"type": "remote",
"url": "http://localhost:7860/mcp"
}
}
}

37
src/PMOCrazyCoder.py Normal file
View File

@@ -0,0 +1,37 @@
from crazy_mcp import Context, mcp
from crazy_mcp.tools import *
# Déclaration d'un outil
@mcp.tool(description="Renvoie le texte reçu")
def echo_texte(texte: str) -> str:
return f"Écho du serveur : {texte} mec"
# Déclaration d'un second outil pour démonstration
@mcp.tool(description="Renvoie le texte en majuscules")
def shout_texte(texte: str) -> str:
return texte.upper()
# Déclaration d'un outil pour additionner deux nombres
@mcp.tool(description="Additionne deux nombres")
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(
transport="http", # Transport réseau HTTP
host="0.0.0.0", # Adresse découte
port=7860, # Port TCP
path="/mcp", # Chemin MCP (par défaut /mcp)
)

View File

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

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)

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

View File

@@ -0,0 +1,51 @@
import json
import re
from ollama import prompt
def structure_to_json(text, schema):
"""
Convert structured text to JSON using Ollama.
Args:
text (str): The structured text to convert
schema (dict): The JSON schema to use for conversion
Returns:
dict: The parsed JSON object
"""
# Format the schema as JSON string
schema_json = json.dumps(schema, indent=2)
# Create the prompt following the pattern from your bash command
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 extra text
Input text:
{text}
Schema:
{schema_json}"""
# Send to Ollama
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
clean_response = response.strip()
while clean_response.startswith("`") or clean_response.startswith("```"):
clean_response = clean_response[1:].strip()
while clean_response.endswith("`") or clean_response.endswith("```"):
clean_response = clean_response[:-1].strip()
# Try to parse the response as JSON
try:
return json.loads(clean_response)
except json.JSONDecodeError:
# If parsing fails, return the raw response
return {"error": "Failed to parse JSON", "raw_response": clean_response}

39
src/jujutsu/__init__.py Normal file
View File

@@ -0,0 +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)

103
src/jujutsu/command.py Normal file
View File

@@ -0,0 +1,103 @@
import json
import os
import subprocess
from typing import List, Optional
def jj(
subcommand: str,
options: Optional[List[str]] = None,
args: Optional[List[str]] = None,
template: Optional[str] = None,
working_dir: Optional[str] = None,
) -> str:
"""
Lance une commande jj.
Args:
subcommand: La sous-commande jj (ex: 'log', 'status', 'diff')
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
"""
cmd = ["jj", subcommand]
# Ajoute le template spécifié
if template:
cmd.extend(["--template", template])
if options:
cmd.extend(options)
if args:
cmd.extend(args)
# 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(
f"Commande jj échouée: {' '.join(cmd)}\nstderr: {result.stderr}"
)
return result.stdout
def jj_log(options=None, args=None, template=None, working_dir: Optional[str] = None):
"""
Execute jj log command with optional arguments.
Args:
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, working_dir=working_dir)
def jj_clone(url, destination=None, options=None, working_dir: Optional[str] = None):
"""
Execute jj git clone command.
Args:
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
"""
# 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_args.append(destination)
if options:
cmd_args.extend(options)
# Call the main jj function with subcommand "git clone"
return jj(
"git",
options=["clone", url],
args=destination,
template=None,
working_dir=working_dir,
)

83
src/ollama/__init__.py Normal file
View File

@@ -0,0 +1,83 @@
import os
import requests
# Get Ollama server address from environment variable, default to local address
ollama_addr = os.environ.get("OLLAMA_ADDR", "http://host.docker.internal:11434")
def prompt(model, text, allow_pull=False):
"""
Send a prompt to Ollama server and return the response.
Args:
model (str): The model to use
text (str): The prompt text to send
allow_pull (bool): If True, automatically download the model if not available
Returns:
str: The response from the Ollama server
"""
# Check if model is available
if allow_pull and not is_model_available(model):
# Try to pull the model
pull_result = pull_model(model)
if not pull_result.startswith("Model downloaded successfully"):
return pull_result # Return error from pull
url = f"{ollama_addr}/api/generate"
payload = {"model": model, "prompt": text, "stream": False}
try:
response = requests.post(url, json=payload)
response.raise_for_status()
return response.json()["response"]
except requests.exceptions.RequestException as e:
return f"Error: {str(e)}"
def pull_model(model_name, force=False):
"""
Download a model from Ollama.
Args:
model_name (str): The name of the model to download
force (bool): If True, forces download even if model already exists
Returns:
str: Success message or error message
"""
url = f"{ollama_addr}/api/pull"
payload = {"name": model_name}
if force:
payload["force"] = True
try:
response = requests.post(url, json=payload)
response.raise_for_status()
return "Model downloaded successfully"
except requests.exceptions.RequestException as e:
return f"Error downloading model: {str(e)}"
def is_model_available(model_name):
"""
Check if a model is available in Ollama.
Args:
model_name (str): The name of the model to check
Returns:
bool: True if model is available, False otherwise
"""
url = f"{ollama_addr}/api/tags"
try:
response = requests.get(url)
response.raise_for_status()
models = response.json().get("models", [])
return any(model.get("name") == model_name for model in models)
except requests.exceptions.RequestException:
return False

139
src/sandbox/__init__.py Normal file
View File

@@ -0,0 +1,139 @@
import os
import re
import shutil
import subprocess
from pathlib import Path
from jujutsu.command import jj_clone
__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."""
return __sandbox_root__
def install_project(url, project_name=None):
"""
Clone a Git project into the sandbox root directory.
Args:
url: The Git repository URL to clone
project_name: Optional project name (if None, derived from URL)
Returns:
str: The name of the cloned project
"""
# If no project name is provided, derive it from the URL
if project_name is 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)
# 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
def remove_project(project_name):
"""
Remove a project from the sandbox.
Args:
project_name (str): The name of the project to remove
Returns:
bool: True if the project was removed successfully, False otherwise
"""
project_path = os.path.join(__sandbox_root__, project_name)
try:
if os.path.exists(project_path):
shutil.rmtree(project_path)
return True
return False
except (OSError, PermissionError):
return False
def sandboxed_project_list():
"""
List all projects available in the sandbox.
Returns:
list: A list of project names (strings)
"""
projects = []
try:
with os.scandir(__sandbox_root__) as entries:
for entry in entries:
if entry.is_dir():
projects.append(entry.name)
except (OSError, PermissionError):
# Return empty list if we can't access the directory
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]

View File

7
test.py Normal file
View File

@@ -0,0 +1,7 @@
import requests
url = "http://localhost:6666" # point d'entrée principal
payload = {"tool": "echo", "input": {"text": "Bonjour MCP!"}}
resp = requests.post(url, json=payload)
print(resp.json()) # {"echo": "Bonjour MCP!"} attendu

View File

@@ -0,0 +1,33 @@
#!/bin/bash
URL="https://go.dev/dl/"
OS=$(uname -a | awk '{print $1}')
ARCH=$(uname -m)
if [[ "$ARCH" == "x86_64" ]] ; then
ARCH="amd64"
fi
if [[ "$ARCH" == "aarch64" ]] ; then
ARCH="arm64"
fi
GOFILE=$(curl "$URL" \
| grep 'class="download"' \
| grep "\.tar\.gz" \
| sed -E 's@^.*/dl/(go[1-9].+\.tar\.gz)".*$@\1@' \
| grep -i "$OS" \
| grep -i "$ARCH" \
| head -1)
GOURL=$(curl "${URL}${GOFILE}" \
| sed -E 's@^.*href="(.*\.tar\.gz)".*$@\1@')
echo "Install GO from : $GOURL" 1>&2
mkdir -p /usr/local
cd /usr/local
curl "$GOURL" \
| tar zxf -