summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/mml_composemusic_mcp/server.py323
-rw-r--r--src/mml_composemusic_mcp/synthesizer.py4
2 files changed, 217 insertions, 110 deletions
diff --git a/src/mml_composemusic_mcp/server.py b/src/mml_composemusic_mcp/server.py
index e3612da..f0c8bd1 100644
--- a/src/mml_composemusic_mcp/server.py
+++ b/src/mml_composemusic_mcp/server.py
@@ -1,10 +1,15 @@
# 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 .ir import ErrorCode, ErrorDetail, ErrorPhase, NoteSequence
from .lexer import tokenize
@@ -19,6 +24,15 @@ mcp = FastMCP("mml-composemusic-mcp")
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)
@@ -49,6 +63,33 @@ def _split_errors(errors: list[ErrorDetail]) -> tuple[list[dict], list[dict]]:
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)
@@ -82,104 +123,124 @@ def _parse_mml(mml: str, mode: str) -> tuple[dict | None, list[ErrorDetail]]:
return (note_sequence.to_dict() if note_sequence else None), errors
-@mcp.tool()
-def compose_mml(
+def _execute_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 retro chiptune-style MML."""
+) -> 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 {"mml": mml_text, "description": description}
+ return ToolExecution({"mml": mml_text, "description": description})
if action in ("compose", "validate"):
if not mml or not mode:
- return {
- "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": [],
+ 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 {
- "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 [],
- }
+ 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 {
- "success": False,
- "wav_path": None,
- "duration_sec": 0,
- "note_sequence": note_sequence_dict,
- "validation": {"errors": errors_list, "warnings": warnings_list},
- }
+ 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 as exc:
- return {
- "success": False,
- "wav_path": 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": f"音声合成中にエラーが発生しました: {exc}",
- "severity": "error",
- "hint": "MMLの内容を確認の上、再度お試しください。問題が続く場合は、短いMMLから試してください。",
- }
- ],
- "warnings": warnings_list,
+ 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 {
- "success": False,
- "wav_path": None,
- "duration_sec": 0,
- "note_sequence": note_sequence_dict,
- "validation": {"errors": errors_list, "warnings": warnings_list},
- }
+ 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)
@@ -187,65 +248,111 @@ def compose_mml(
if not _is_error(errors):
try:
(output_dir / "output.mml").write_text(mml, encoding="utf-8")
- except Exception as exc:
+ except Exception:
errors.append(
ErrorDetail(
code=ErrorCode.RUNTIME_MML_WRITE_FAILED,
phase=ErrorPhase.RUNTIME,
line=0,
column=0,
- message=f"MMLファイルの出力に失敗しました: {exc}",
+ message="MMLファイルの出力に失敗しました。",
severity="error",
hint="出力先を確認してください。",
)
)
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,
+ 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},
}
- except Exception as exc:
- return {
- "success": False,
- "wav_path": None,
- "duration_sec": 0,
- "note_sequence": None,
- "validation": {
- "errors": [
- {
- "code": ErrorCode.RUNTIME_INTERNAL_ERROR.value,
- "phase": ErrorPhase.RUNTIME.value,
- "line": 0,
- "column": 0,
- "message": f"内部エラーが発生しました: {exc}",
- "severity": "error",
- "hint": "MMLの内容を確認の上、再度お試しください。",
- }
- ],
- "warnings": [],
- },
- }
+ return ToolExecution(
+ result,
+ is_error=failed,
+ wav_path=None if failed else wav_path,
+ )
+ except Exception:
+ return _internal_error_execution()
- return {
- "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": [],
+ 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:
diff --git a/src/mml_composemusic_mcp/synthesizer.py b/src/mml_composemusic_mcp/synthesizer.py
index a289a1e..7085342 100644
--- a/src/mml_composemusic_mcp/synthesizer.py
+++ b/src/mml_composemusic_mcp/synthesizer.py
@@ -545,14 +545,14 @@ def write_wav(path: Path, wave_data: np.ndarray, sample_rate: int) -> list[Error
wav.setsampwidth(2)
wav.setframerate(sample_rate)
wav.writeframes(pcm.tobytes())
- except Exception as exc:
+ except Exception:
errors.append(
ErrorDetail(
code=ErrorCode.RUNTIME_WAV_WRITE_FAILED,
phase=ErrorPhase.RUNTIME,
line=0,
column=0,
- message=f"WAVファイルの出力に失敗しました: {exc}",
+ message="WAVファイルの出力に失敗しました。",
severity="error",
hint="出力先を確認してください。",
)
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-19 19:42:44 +0000