Add Phase 2: MIDI control with synth definitions, tools, and deps

Introduce MIDI output support via python-rtmidi, synth definition loading
from YAML files, PydanticAI dependency injection, and agent tools for
controlling hardware synths over MIDI CC. Includes Minitaur and TB-03
synth definitions, comprehensive test coverage for all new modules.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pnilan

patchwork/agent.py

@@ -1,4 +1,13 @@
-from pydantic_ai import Agent
+from pydantic_ai import Agent, RunContext
+
+from patchwork.deps import PatchworkDeps
+from patchwork.tools.midi_control import (
+    connect_midi,
+    list_midi_ports,
+    list_synths,
+    send_cc,
+    send_patch,
+)
 
 SYSTEM_PROMPT = """You are Patchwork, an expert synthesizer sound design assistant.
 
@@ -16,8 +25,15 @@
 - 1010 Music Blackbox (sampler/groovebox)
 - Arturia Minibrute 2S (analog, semi-modular)
 
-You will eventually control these synths via MIDI CC messages. For now, describe
-settings conceptually using parameter names and values.
+You have tools to control synths via MIDI:
+- list_midi_ports: Show available MIDI output devices
+- connect_midi: Connect to a MIDI port
+- list_synths: Show loaded synth definitions and their controllable parameters
+- send_cc: Send a single MIDI CC value to a synth parameter
+- send_patch: Send multiple MIDI CC values at once to set a full patch
+
+When the user asks you to set a sound or tweak a parameter, use these tools.
+Always confirm what you sent after a successful tool call.
 
 Tone: conversational but concise. Use musical and technical terminology naturally.
 When describing a patch, be specific enough that someone could recreate it by hand.
@@ -26,5 +42,19 @@
 agent = Agent(
     "anthropic:claude-sonnet-4-6",
     system_prompt=SYSTEM_PROMPT,
+    deps_type=PatchworkDeps,
     defer_model_check=True,
+    tools=[list_midi_ports, connect_midi, list_synths, send_cc, send_patch],
 )
+
+
+@agent.system_prompt
+async def add_synth_context(ctx: RunContext[PatchworkDeps]) -> str:
+    """Append loaded synth definitions to the system prompt at runtime."""
+    if not ctx.deps.synths:
+        return "No synth definitions loaded yet."
+    lines = []
+    for synth in ctx.deps.synths.values():
+        params = ", ".join(synth.cc_map.keys())
+        lines.append(f"- {synth.manufacturer} {synth.name} (ch.{synth.midi_channel}): {params}")
+    return "Loaded synths and their controllable parameters:\n" + "\n".join(lines)

patchwork/cli.py

@@ -4,39 +4,53 @@
 from rich.console import Console
 
 from patchwork.agent import agent
+from patchwork.deps import PatchworkDeps
+from patchwork.midi import MidiConnection
+from patchwork.synth_definitions import load_synth_definitions
 
 console = Console()
 
 
 async def main():
+    midi = MidiConnection()
+    synths = load_synth_definitions()
+    deps = PatchworkDeps(midi=midi, synths=synths)
+
     console.print("[bold]patchwork[/bold] — synth research agent\n")
+    if synths:
+        console.print(f"[dim]loaded {len(synths)} synth(s): {', '.join(s.name for s in synths.values())}[/dim]\n")
+    else:
+        console.print("[dim]no synth definitions found in synths/[/dim]\n")
     message_history = []
 
-    while True:
-        try:
-            user_input = console.input("[bold cyan]patch>[/bold cyan] ")
-        except (KeyboardInterrupt, EOFError):
-            console.print("\n[dim]goodbye[/dim]")
-            break
-
-        if not user_input.strip():
-            continue
-
-        if user_input.strip().lower() in ("quit", "exit"):
-            console.print("[dim]goodbye[/dim]")
-            break
-
-        try:
-            async with agent.run_stream(
-                user_input, message_history=message_history
-            ) as result:
-                async for chunk in result.stream_text(delta=True):
-                    console.print(chunk, end="", markup=False, highlight=False)
-                console.print()  # newline after stream
-
-            message_history = result.all_messages()
-        except Exception as e:
-            console.print(f"\n[bold red]error:[/bold red] {e}")
+    try:
+        while True:
+            try:
+                user_input = console.input("[bold cyan]patch>[/bold cyan] ")
+            except (KeyboardInterrupt, EOFError):
+                console.print("\n[dim]goodbye[/dim]")
+                break
+
+            if not user_input.strip():
+                continue
+
+            if user_input.strip().lower() in ("quit", "exit"):
+                console.print("[dim]goodbye[/dim]")
+                break
+
+            try:
+                async with agent.run_stream(
+                    user_input, message_history=message_history, deps=deps
+                ) as result:
+                    async for chunk in result.stream_text(delta=True):
+                        console.print(chunk, end="", markup=False, highlight=False)
+                    console.print()  # newline after stream
+
+                message_history = result.all_messages()
+            except Exception as e:
+                console.print(f"\n[bold red]error:[/bold red] {e}")
+    finally:
+        midi.close()
 
 
 def main_cli():

patchwork/deps.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+from patchwork.midi import MidiConnection
+from patchwork.synth_definitions import SynthDefinition
+
+
+@dataclass
+class PatchworkDeps:
+    midi: MidiConnection
+    synths: dict[str, SynthDefinition]

patchwork/midi.py

@@ -0,0 +1,55 @@
+import rtmidi
+
+
+class MidiConnection:
+    def __init__(self):
+        self._out: rtmidi.MidiOut | None = None
+        self._port_name: str | None = None
+
+    @property
+    def is_connected(self) -> bool:
+        return self._out is not None
+
+    @property
+    def port_name(self) -> str | None:
+        return self._port_name
+
+    def list_ports(self) -> list[str]:
+        """List available MIDI output port names."""
+        out = rtmidi.MidiOut()
+        ports = out.get_ports()
+        del out
+        return ports
+
+    def open(self, port_index: int = 0) -> str:
+        """Open a MIDI output port by index. Returns the port name."""
+        self._out = rtmidi.MidiOut()
+        ports = self._out.get_ports()
+        if not ports:
+            raise RuntimeError("No MIDI output ports available")
+        if port_index >= len(ports):
+            raise ValueError(f"Port index {port_index} out of range (0-{len(ports) - 1})")
+        self._out.open_port(port_index)
+        self._port_name = ports[port_index]
+        return self._port_name
+
+    def close(self):
+        """Close the current MIDI connection."""
+        if self._out is not None:
+            self._out.close_port()
+            del self._out
+            self._out = None
+            self._port_name = None
+
+    def send_cc(self, channel: int, cc_number: int, value: int):
+        """Send a MIDI CC message. Channel is 1-16 (converted to 0-15 internally)."""
+        if self._out is None:
+            raise RuntimeError("MIDI port not open — call open() first")
+        if not (1 <= channel <= 16):
+            raise ValueError(f"MIDI channel must be 1-16, got {channel}")
+        if not (0 <= cc_number <= 127):
+            raise ValueError(f"CC number must be 0-127, got {cc_number}")
+        if not (0 <= value <= 127):
+            raise ValueError(f"CC value must be 0-127, got {value}")
+        status = 0xB0 | (channel - 1)
+        self._out.send_message([status, cc_number, value])

patchwork/synth_definitions.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, Field, model_validator
+
+
+class CCParameter(BaseModel):
+    cc: int = Field(ge=0, le=127)
+    value_range: tuple[int, int] = (0, 127)
+    notes: str | None = None
+
+    @model_validator(mode="after")
+    def check_range_order(self) -> "CCParameter":
+        low, high = self.value_range
+        if low > high:
+            raise ValueError(f"value_range low ({low}) must be <= high ({high})")
+        return self
+
+
+class SynthDefinition(BaseModel):
+    name: str
+    manufacturer: str
+    midi_channel: int = Field(ge=1, le=16)
+    cc_map: dict[str, CCParameter]
+
+
+def load_synth_definitions(synths_dir: Path = Path("synths")) -> dict[str, SynthDefinition]:
+    """Load and validate all synth YAML files from the given directory."""
+    definitions: dict[str, SynthDefinition] = {}
+    if not synths_dir.exists():
+        return definitions
+    yaml_files = sorted([*synths_dir.glob("*.yaml"), *synths_dir.glob("*.yml")])
+    for yaml_file in yaml_files:
+        data = yaml.safe_load(yaml_file.read_text())
+        synth = SynthDefinition(**data)
+        key = synth.name.lower()
+        if key in definitions:
+            raise ValueError(
+                f"Duplicate synth name '{synth.name}' "
+                f"(from {yaml_file.name} and existing definition)"
+            )
+        definitions[key] = synth
+    return definitions

patchwork/tools/__init__.py

@@ -0,0 +1,110 @@
+from pydantic_ai import RunContext
+
+from patchwork.deps import PatchworkDeps
+
+
+async def list_midi_ports(ctx: RunContext[PatchworkDeps]) -> str:
+    """List available MIDI output ports."""
+    ports = ctx.deps.midi.list_ports()
+    if not ports:
+        return "No MIDI output ports found. Is your MIDI interface connected?"
+    lines = [f"{i}: {name}" for i, name in enumerate(ports)]
+    return "Available MIDI ports:\n" + "\n".join(lines)
+
+
+async def send_cc(
+    ctx: RunContext[PatchworkDeps],
+    synth: str,
+    parameter: str,
+    value: int,
+) -> str:
+    """Send a MIDI CC value to a parameter on a synth.
+
+    Args:
+        synth: Synth name (e.g. "minitaur", "tb-03")
+        parameter: Parameter name from the synth's CC map (e.g. "filter_cutoff")
+        value: CC value to send (0-127)
+    """
+    synth_def = ctx.deps.synths.get(synth.lower())
+    if synth_def is None:
+        available = ", ".join(ctx.deps.synths.keys())
+        return f"Unknown synth '{synth}'. Available: {available}"
+
+    param_key = parameter.lower().replace(" ", "_")
+    param = synth_def.cc_map.get(param_key)
+    if param is None:
+        available = ", ".join(synth_def.cc_map.keys())
+        return f"Unknown parameter '{parameter}' for {synth_def.name}. Available: {available}"
+
+    low, high = param.value_range
+    if not (low <= value <= high):
+        return f"Value {value} out of range for {parameter} ({low}-{high})"
+
+    if not ctx.deps.midi.is_connected:
+        return f"MIDI not connected. Use list_midi_ports to see available ports, then ask me to connect."
+
+    ctx.deps.midi.send_cc(synth_def.midi_channel, param.cc, value)
+    return f"Sent CC {param.cc} = {value} to {synth_def.name} ch.{synth_def.midi_channel} ({parameter})"
+
+
+async def send_patch(
+    ctx: RunContext[PatchworkDeps],
+    synth: str,
+    settings: dict[str, int],
+) -> str:
+    """Send multiple CC values at once to set a full patch on a synth.
+
+    Args:
+        synth: Synth name (e.g. "minitaur", "tb-03")
+        settings: Dict of parameter name to CC value (e.g. {"filter_cutoff": 64, "resonance": 80})
+    """
+    synth_def = ctx.deps.synths.get(synth.lower())
+    if synth_def is None:
+        available = ", ".join(ctx.deps.synths.keys())
+        return f"Unknown synth '{synth}'. Available: {available}"
+
+    if not ctx.deps.midi.is_connected:
+        return f"MIDI not connected. Use list_midi_ports to see available ports, then ask me to connect."
+
+    results = []
+    for param_name, value in settings.items():
+        param_key = param_name.lower().replace(" ", "_")
+        param = synth_def.cc_map.get(param_key)
+        if param is None:
+            results.append(f"  ✗ Unknown parameter '{param_name}'")
+            continue
+        low, high = param.value_range
+        if not (low <= value <= high):
+            results.append(f"  ✗ {param_name}: value {value} out of range ({low}-{high})")
+            continue
+        ctx.deps.midi.send_cc(synth_def.midi_channel, param.cc, value)
+        results.append(f"  ✓ {param_name} = {value} (CC {param.cc})")
+
+    return f"Patch sent to {synth_def.name} ch.{synth_def.midi_channel}:\n" + "\n".join(results)
+
+
+async def list_synths(ctx: RunContext[PatchworkDeps]) -> str:
+    """List all loaded synth definitions and their controllable parameters."""
+    if not ctx.deps.synths:
+        return "No synth definitions loaded. Add YAML files to the synths/ directory."
+    lines = []
+    for synth in ctx.deps.synths.values():
+        params = ", ".join(synth.cc_map.keys())
+        lines.append(f"- {synth.manufacturer} {synth.name} (ch.{synth.midi_channel}): {params}")
+    return "Loaded synths:\n" + "\n".join(lines)
+
+
+async def connect_midi(
+    ctx: RunContext[PatchworkDeps],
+    port_index: int = 0,
+) -> str:
+    """Connect to a MIDI output port by index. Use list_midi_ports first to see available ports.
+
+    Args:
+        port_index: Index of the MIDI port to connect to (default: 0, the first port)
+    """
+    try:
+        port_name = ctx.deps.midi.open(port_index)
+        return f"Connected to MIDI port: {port_name}"
+    except (RuntimeError, ValueError) as e:
+        return str(e)

patchwork/tools/midi_control.py

@@ -8,6 +8,8 @@ dependencies = [
     "pydantic-ai",
     "rich",
     "python-dotenv",
+    "python-rtmidi",
+    "pyyaml",
 ]
 
 [project.scripts]

pyproject.toml

@@ -0,0 +1,30 @@
+name: Minitaur
+manufacturer: Moog
+midi_channel: 2
+cc_map:
+  filter_cutoff:
+    cc: 22
+  filter_resonance:
+    cc: 23
+  filter_eg_amount:
+    cc: 24
+  osc2_frequency:
+    cc: 18
+  osc_mix:
+    cc: 15
+  vca_volume:
+    cc: 7
+  eg_attack:
+    cc: 105
+  eg_decay:
+    cc: 106
+  eg_sustain:
+    cc: 107
+  eg_release:
+    cc: 108
+  glide_rate:
+    cc: 5
+  legato_mode:
+    cc: 114
+    value_range: [0, 127]
+    notes: "0=off, 127=on"