From ca35520338057512a71acd42bb116ffbb82cc192 Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Sat, 11 Jul 2026 19:02:14 +0900 Subject: feat: implement Famicom-style MML composer MCP server - Add FastMCP server with compose_mml tool - Support ppmck and pyxel MML syntax - Add lexer and parsers for both formats - Add NES APU-style synthesizer (pulse/triangle/noise) to WAV - Add MML templates (basic/melody/chord/drum/empty) for both modes - Add tests for lexer, parser, and server - Support stdio, http, sse, streamable-http transports - Update README, pyproject.toml, and .gitignore --- src/mml_composemusic_mcp/server.py | 254 +++++++++++++++++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 src/mml_composemusic_mcp/server.py (limited to 'src/mml_composemusic_mcp/server.py') diff --git a/src/mml_composemusic_mcp/server.py b/src/mml_composemusic_mcp/server.py new file mode 100644 index 0000000..1205b70 --- /dev/null +++ b/src/mml_composemusic_mcp/server.py @@ -0,0 +1,254 @@ +# MCP server exposing compose_mml tool. + +import argparse +from pathlib import Path + +from fastmcp import FastMCP + +from .ir import ErrorCode, ErrorDetail, NoteSequence +from .lexer import tokenize +from .parser_ppmck import parse_ppmck +from .parser_pyxel import parse_pyxel +from .synthesizer import build_channel_summary, synthesize, write_wav +from .templates import get_template + +mcp = FastMCP("mml-composemusic-mcp") +OUTPUT_DIR = Path("./data") + + +def _is_error(errors: list[ErrorDetail]) -> bool: + return any(e.severity == "error" for e in errors) + + +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 _parse_mml(mml: str, mode: str) -> tuple[dict, list[ErrorDetail]]: + tokens = tokenize(mml, mode) + if mode == "ppmck": + note_sequence, errors = parse_ppmck(mml, tokens) + elif mode == "pyxel": + note_sequence, errors = parse_pyxel(mml, tokens) + else: + errors = [ + ErrorDetail( + code=ErrorCode.SYNTAX_INVALID_TOKEN, + line=0, + column=0, + message=f"未知のモード '{mode}' です。", + severity="error", + hint="mode は 'ppmck' または 'pyxel' を指定してください。", + ) + ] + note_sequence = None + return (note_sequence.to_dict() if note_sequence else None), errors + + +@mcp.tool() +def compose_mml( + action: str, + mml: str = "", + mode: str = "", + template: str = "basic", + sample_rate: int = 44100, + normalize: bool = True, +) -> dict: + """Compose, validate, or generate templates for Famicom-style MML.""" + if action == "template": + if mode not in ("ppmck", "pyxel"): + mode = "ppmck" + if template not in ("basic", "melody", "chord", "drum", "empty"): + template = "basic" + mml_text, description = get_template(mode, template) + return {"mml": mml_text, "description": description} + + if action in ("compose", "validate"): + if not mml or not mode: + return { + "success": False, + "valid": False, + "errors": [ + { + "code": ErrorCode.SYNTAX_INVALID_TOKEN.value, + "line": 0, + "column": 0, + "message": "mml と mode は compose/validate の必須パラメータです。", + "severity": "error", + "hint": "mml に MML 文字列、mode に 'ppmck' または 'pyxel' を指定してください。", + } + ], + "warnings": [], + } + note_sequence_dict, errors = _parse_mml(mml, mode) + errors_list, warnings_list = _split_errors(errors) + if action == "validate": + return { + "valid": not _is_error(errors), + "errors": errors_list, + "note_sequence": note_sequence_dict, + "channel_summary": build_channel_summary(note_sequence_dict) + if note_sequence_dict + else [], + } + # compose + if _is_error(errors): + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + + ns = _dict_to_note_sequence(note_sequence_dict) + try: + wave_data, duration, synth_errors = synthesize( + ns, mode, sample_rate, normalize + ) + except Exception as exc: + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": { + "errors": [ + { + "code": ErrorCode.SYSTEM_SYNTHESIS_FAILED.value, + "line": 0, + "column": 0, + "message": f"音声合成中にエラーが発生しました: {exc}", + "severity": "error", + "hint": "MMLの内容を確認の上、再度お試しください。問題が続く場合は、短いMMLから試してください。", + } + ], + "warnings": warnings_list, + }, + } + errors.extend(synth_errors) + errors_list, warnings_list = _split_errors(errors) + if _is_error(errors): + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + wav_path = OUTPUT_DIR / "output.wav" + wav_errors = write_wav(wav_path, wave_data, sample_rate) + errors.extend(wav_errors) + errors_list, warnings_list = _split_errors(errors) + return { + "success": not _is_error(errors), + "wav_path": str(wav_path) if not _is_error(errors) else None, + "duration_sec": duration, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + + return { + "success": False, + "errors": [ + { + "code": ErrorCode.SYNTAX_INVALID_TOKEN.value, + "line": 0, + "column": 0, + "message": f"未知の action '{action}' です。", + "severity": "error", + "hint": "action は 'compose', 'validate', 'template' のいずれかを指定してください。", + } + ], + "warnings": [], + } + + +def _dict_to_note_sequence(data: dict) -> NoteSequence: + from .ir import ( + ChannelSequence, + DutyEvent, + EnvelopeEvent, + GlideEvent, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, + VibratoEvent, + VolumeEvent, + ) + + ns = NoteSequence( + version=data["version"], + bpm=data["bpm"], + ticks_per_quarter=data["ticks_per_quarter"], + 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 == "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) + mcp.run(transport=args.transport, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() -- cgit v1.2.3