# MCP server exposing compose_mml tool. import argparse import json from dataclasses import dataclass from datetime import datetime, timedelta from pathlib import Path from fastmcp import FastMCP from fastmcp.tools import ToolResult from fastmcp.utilities.types import Audio from mcp.types import TextContent from . import __version__ from .ir import ErrorCode, ErrorDetail, ErrorPhase, NoteSequence from .lexer import tokenize from .parser_ppmck import parse_ppmck from .parser_pyxel import parse_pyxel from .semantic_ppmck import analyze_ppmck from .semantic_pyxel import analyze_pyxel from .synthesizer import build_channel_summary, synthesize, write_wav from .templates import TEMPLATES, get_template mcp = FastMCP("mml-composemusic-mcp", version=__version__) OUTPUT_DIR = Path("./data") @dataclass(frozen=True) class ToolExecution: """Internal result kept separate from the public MCP payload.""" result: dict is_error: bool = False wav_path: Path | None = None def _is_error(errors: list[ErrorDetail]) -> bool: return any(e.severity == "error" for e in errors) def _create_output_directory() -> Path: """Create and return a unique timestamped output directory. Format: YYYYMMDD_HHMMSS_mmm If the millisecond is already in use, advance the timestamp until a free directory can be created. Exclusive creation also handles concurrent calls. """ now = datetime.now() for offset_ms in range(1000): candidate_time = now + timedelta(milliseconds=offset_ms) timestamp = candidate_time.strftime("%Y%m%d_%H%M%S_%f")[:-3] output_dir = OUTPUT_DIR / timestamp try: output_dir.mkdir(parents=True, exist_ok=False) return output_dir except FileExistsError: continue raise RuntimeError("一意な出力ディレクトリを作成できませんでした。") def _split_errors(errors: list[ErrorDetail]) -> tuple[list[dict], list[dict]]: errs = [e.to_dict() for e in errors if e.severity == "error"] warns = [e.to_dict() for e in errors if e.severity == "warning"] return errs, warns def _internal_error_execution() -> ToolExecution: """Build a path-free MCP tool error for unexpected failures.""" return ToolExecution( { "success": False, "audio": None, "duration_sec": 0, "note_sequence": None, "validation": { "errors": [ { "code": ErrorCode.RUNTIME_INTERNAL_ERROR.value, "phase": ErrorPhase.RUNTIME.value, "line": 0, "column": 0, "message": "内部エラーが発生しました。", "severity": "error", "hint": "MMLの内容を確認の上、再度お試しください。", } ], "warnings": [], }, }, is_error=True, ) def _parse_mml(mml: str, mode: str) -> tuple[dict | None, list[ErrorDetail]]: """Run lexer, parser, and semantic analyzer for the given mode.""" tokens = tokenize(mml, mode) if mode == "ppmck": program, errors = parse_ppmck(mml, tokens) if not _is_error(errors): note_sequence, semantic_errors = analyze_ppmck(mml, program) errors.extend(semantic_errors) elif mode == "pyxel": program, errors = parse_pyxel(mml, tokens) if not _is_error(errors): note_sequence, semantic_errors = analyze_pyxel(mml, program) errors.extend(semantic_errors) else: errors = [ ErrorDetail( code=ErrorCode.VALIDATION_INVALID_MODE, phase=ErrorPhase.API, line=0, column=0, message=f"未知のモード '{mode}' です。", severity="error", hint="mode は 'ppmck' または 'pyxel' を指定してください。", ) ] note_sequence = None if _is_error(errors): note_sequence = None return (note_sequence.to_dict() if note_sequence else None), errors def _execute_compose_mml( action: str, mml: str = "", mode: str = "", template: str = "basic", sample_rate: int = 44100, normalize: bool = True, ) -> ToolExecution: """Execute a request without exposing internal file paths.""" if action == "template": if mode not in ("ppmck", "pyxel"): mode = "ppmck" if template not in TEMPLATES[mode]: template = "basic" mml_text, description = get_template(mode, template) return ToolExecution({"mml": mml_text, "description": description}) if action in ("compose", "validate"): if not mml or not mode: return ToolExecution( { "success": False, "valid": False, "note_sequence": None, "validation": { "errors": [ { "code": ErrorCode.VALIDATION_MISSING_PARAMETER.value, "phase": ErrorPhase.API.value, "line": 0, "column": 0, "message": "mml と mode は compose/validate の必須パラメータです。", "severity": "error", "hint": "mml に MML 文字列、mode に 'ppmck' または 'pyxel' を指定してください。", } ], "warnings": [], }, }, is_error=True, ) try: note_sequence_dict, errors = _parse_mml(mml, mode) errors_list, warnings_list = _split_errors(errors) if action == "validate": return ToolExecution( { "valid": not _is_error(errors), "errors": errors_list, "warnings": warnings_list, "note_sequence": note_sequence_dict, "channel_summary": build_channel_summary(note_sequence_dict) if note_sequence_dict else [], }, is_error=mode not in ("ppmck", "pyxel"), ) # compose if _is_error(errors): return ToolExecution( { "success": False, "audio": None, "duration_sec": 0, "note_sequence": note_sequence_dict, "validation": { "errors": errors_list, "warnings": warnings_list, }, }, is_error=True, ) ns = _dict_to_note_sequence(note_sequence_dict) try: wave_data, duration, synth_errors = synthesize( ns, mode, sample_rate, normalize ) except Exception: return ToolExecution( { "success": False, "audio": None, "duration_sec": 0, "note_sequence": note_sequence_dict, "validation": { "errors": [ { "code": ErrorCode.RUNTIME_SYNTHESIS_FAILED.value, "phase": ErrorPhase.RUNTIME.value, "line": 0, "column": 0, "message": "音声合成中にエラーが発生しました。", "severity": "error", "hint": "MMLの内容を確認の上、再度お試しください。問題が続く場合は、短いMMLから試してください。", } ], "warnings": warnings_list, }, }, is_error=True, ) errors.extend(synth_errors) errors_list, warnings_list = _split_errors(errors) if _is_error(errors): return ToolExecution( { "success": False, "audio": None, "duration_sec": 0, "note_sequence": note_sequence_dict, "validation": { "errors": errors_list, "warnings": warnings_list, }, }, is_error=True, ) output_dir = _create_output_directory() wav_path = output_dir / "output.wav" wav_errors = write_wav(wav_path, wave_data, sample_rate) errors.extend(wav_errors) if not _is_error(errors): try: (output_dir / "output.mml").write_text(mml, encoding="utf-8") except Exception: errors.append( ErrorDetail( code=ErrorCode.RUNTIME_MML_WRITE_FAILED, phase=ErrorPhase.RUNTIME, line=0, column=0, message="MMLファイルの出力に失敗しました。", severity="error", hint="出力先を確認してください。", ) ) errors_list, warnings_list = _split_errors(errors) failed = _is_error(errors) result = { "success": not failed, "audio": None if failed else { "mime_type": "audio/wav", "size_bytes": wav_path.stat().st_size, "sample_rate": sample_rate, }, "duration_sec": duration, "note_sequence": note_sequence_dict, "validation": {"errors": errors_list, "warnings": warnings_list}, } return ToolExecution( result, is_error=failed, wav_path=None if failed else wav_path, ) except Exception: return _internal_error_execution() return ToolExecution( { "success": False, "validation": { "errors": [ { "code": ErrorCode.VALIDATION_INVALID_ACTION.value, "phase": ErrorPhase.API.value, "line": 0, "column": 0, "message": f"未知の action '{action}' です。", "severity": "error", "hint": "action は 'compose', 'validate', 'template' のいずれかを指定してください。", } ], "warnings": [], }, }, is_error=True, ) def _build_tool_result(execution: ToolExecution) -> ToolResult: """Convert an internal execution result to the public MCP response.""" result = execution.result if "wav_path" in result: raise ValueError("Internal paths must not be included in MCP results") if "success" in result and result["success"] is execution.is_error: raise ValueError("success and is_error are inconsistent") if execution.is_error and execution.wav_path is not None: raise ValueError("Error results must not contain audio") content = [ TextContent( type="text", text=json.dumps(result, ensure_ascii=False), ) ] if execution.wav_path is not None: content.append(Audio(path=execution.wav_path).to_audio_content()) return ToolResult( content=content, structured_content=result, is_error=execution.is_error, ) @mcp.tool(name="compose_mml", output_schema=None) def compose_mml( action: str, mml: str = "", mode: str = "", template: str = "basic", sample_rate: int = 44100, normalize: bool = True, ) -> ToolResult: """Compose, validate, or generate templates for retro chiptune-style MML.""" try: execution = _execute_compose_mml( action=action, mml=mml, mode=mode, template=template, sample_rate=sample_rate, normalize=normalize, ) return _build_tool_result(execution) except Exception: return _build_tool_result(_internal_error_execution()) def _dict_to_note_sequence(data: dict) -> NoteSequence: from .ir import ( ChannelSequence, DetuneEvent, DutyEnvelopeEvent, DutyEvent, EnvelopeEvent, GlideEvent, LfoEvent, NoteEnvEvent, NoteEvent, NoteSequence, PitchEnvEvent, QuantizeEvent, RelativeVolumeEvent, RepeatEvent, RestEvent, SweepEvent, TempoEvent, VibratoEvent, VolumeEnvelopeEvent, VolumeEvent, ) ns = NoteSequence( version=data["version"], bpm=data["bpm"], ticks_per_quarter=data["ticks_per_quarter"], mode=data.get("mode", "ppmck"), definitions=data.get("definitions", {}), channels={}, ) for name, ch_data in data["channels"].items(): ch = ChannelSequence( channel_type=ch_data["channel_type"], events=[], total_ticks=ch_data["total_ticks"], ) for ev in ch_data["events"]: ev_type = ev["type"] if ev_type == "note": ch.events.append(NoteEvent(**ev)) elif ev_type == "rest": ch.events.append(RestEvent(**ev)) elif ev_type == "volume": ch.events.append(VolumeEvent(**ev)) elif ev_type == "duty": ch.events.append(DutyEvent(**ev)) elif ev_type == "tempo": ch.events.append(TempoEvent(**ev)) elif ev_type == "repeat": ch.events.append(RepeatEvent(**ev)) elif ev_type == "envelope": ch.events.append(EnvelopeEvent(**ev)) elif ev_type == "quantize": ch.events.append(QuantizeEvent(**ev)) elif ev_type == "detune": ch.events.append(DetuneEvent(**ev)) elif ev_type == "sweep": ch.events.append(SweepEvent(**ev)) elif ev_type == "rel_volume": ch.events.append(RelativeVolumeEvent(**ev)) elif ev_type == "vol_envelope": ch.events.append(VolumeEnvelopeEvent(**ev)) elif ev_type == "duty_envelope": ch.events.append(DutyEnvelopeEvent(**ev)) elif ev_type == "lfo": ch.events.append(LfoEvent(**ev)) elif ev_type == "pitch_envelope": ch.events.append(PitchEnvEvent(**ev)) elif ev_type == "note_envelope": ch.events.append(NoteEnvEvent(**ev)) elif ev_type == "vibrato": ch.events.append(VibratoEvent(**ev)) elif ev_type == "glide": ch.events.append(GlideEvent(**ev)) ns.channels[name] = ch return ns def main() -> None: parser = argparse.ArgumentParser(description="MML Compose MCP Server") parser.add_argument( "--output-dir", type=str, default="./data", help="Directory to store generated WAV files (default: ./data)", ) parser.add_argument( "--transport", type=str, default="stdio", choices=["stdio", "http", "sse", "streamable-http"], help="MCP transport (default: stdio)", ) parser.add_argument( "--host", type=str, default="127.0.0.1", help="Host to bind for HTTP-based transports (default: 127.0.0.1)", ) parser.add_argument( "--port", type=int, default=8080, help="Port to bind for HTTP-based transports (default: 8080)", ) args = parser.parse_args() global OUTPUT_DIR OUTPUT_DIR = Path(args.output_dir) if args.transport == "stdio": mcp.run(transport="stdio") else: mcp.run( transport=args.transport, host=args.host, port=args.port, ) if __name__ == "__main__": main()