summaryrefslogtreecommitdiff
path: root/src/mml_composemusic_mcp/server.py
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-07-11 19:02:14 +0900
committertanaka takashi <isazaka@gmail.com>2026-07-11 19:02:14 +0900
commitca35520338057512a71acd42bb116ffbb82cc192 (patch)
tree9faa2428aeb71b62497f21ceec93b99ef7f2a5ac /src/mml_composemusic_mcp/server.py
parent7dca1616b8c891f5e140c7267323d1fcdde6741a (diff)
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
Diffstat (limited to 'src/mml_composemusic_mcp/server.py')
-rw-r--r--src/mml_composemusic_mcp/server.py254
1 files changed, 254 insertions, 0 deletions
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()