Budowa serwera MCP

Omni ma własny serwer MCP, omni-mcp. Ta strona jest dla sytuacji, gdy chcesz inny: serwer społeczności w Twoim ulubionym języku, taki, który wystawia tylko część API, z własnymi promptami albo z dodatkowym przepływem pracy. Nie potrzebujesz kodu źródłowego Omni, tylko jego API. Przykładowy kod na tej stronie jest na licencji MIT, więc możesz od niego zacząć i opublikować to, co zbudujesz.

Co serwer musi robić

Serwer MCP dla Omni to klient OmniDAW API, który po drugiej stronie mówi MCP do asystenta AI. Omni wykonuje już całą pracę (tory, klipy, teoria, mierniki, cofanie); Twój program łączy obie strony:

1. Znajduje Omni przez plik endpointu i otwiera jedno połączenie TCP.
2. Wysyła api.hello z tokenem.
3. Czyta katalog przez api.methods.
4. Opisuje asystentowi każdą metodę jako narzędzie MCP.
5. Gdy asystent wywoła narzędzie, wysyła odpowiadające mu wywołanie API i zwraca wynik albo błąd jako błąd narzędzia.

To wszystko. Katalog zawiera już tekst potrzebny modelowi, więc nie ma nic do pisania dla poszczególnych metod.

Od katalogu do narzędzia

Każdy wpis api.methods przekłada się na jedno narzędzie MCP, pole po polu:

Listę buduj z api.methods przy starcie serwera zamiast pisać ją ręcznie. Wtedy nowsze Omni przynosi nowe metody do Twojego serwera bez wydania nowej wersji.

Pole effect jest też Twoim modelem uprawnień. Klient może uruchamiać narzędzia read bez pytania, pytać przed edit i pytać wyraźnie przed destructive. Ponieważ każda edycja to jeden krok cofania w Omni, nawet błędna edycja jest tania do cofnięcia, a edit_undo samo jest narzędziem.

Kompletny serwer w Pythonie

Poniższy serwer zamienia każdą metodę w narzędzie. Używa klienta Pythona (omnidaw.py, zapisany w tym samym folderze co serwer) i oficjalnego SDK MCP (pip install mcp). Został przetestowany oficjalnym klientem SDK przy działającym Omni: wszystkie 102 narzędzia zostały wylistowane z adnotacjami, odczyty i wywołania narzędzi działały, a źle napisany parametr wrócił jako czytelny błąd narzędzia. Pobierz serwer albo wszystkie przykłady jako jeden zip.

from __future__ import annotations

import asyncio
import json
import sys
import time

from mcp import types
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server

from omnidaw import Omni

INSTRUCTIONS = (
    "Tools drive the project open in Omni, a DAW. Start with session_info and "
    "tracks_list. Time is in beats, levels in dB, pitch as MIDI numbers. Every "
    "edit is one undo step: edit_undo takes back the newest. Long work "
    "(analysis_capture, export_render) returns a job: call jobs_wait with its id."
)

WAIT_FOR_OMNI_SECONDS = 30

_omni: Omni | None = None
#: MCP tool name -> catalog method, filled once at start-up.
_catalog: dict[str, dict] = {}


def connect() -> Omni:
    """The connection to Omni, opened on first use and again after it drops."""
    global _omni
    if _omni is None:
        _omni = Omni("omni-mcp-example")
    return _omni


def tool_of(method: dict) -> types.Tool:
    """One catalog method as an MCP tool."""
    effect = method["effect"]
    return types.Tool(
        name=method["name"].replace(".", "_"),
        description=method["summary"],
        input_schema=method["params"],
        annotations=types.ToolAnnotations(
            title=method["title"],
            read_only_hint=effect == "read",
            destructive_hint=effect == "destructive",
            idempotent_hint=method["idempotent"],
            open_world_hint=False,
        ),
    )


def run_call(method: str, arguments: dict) -> str:
    """The blocking part of a tool call; runs in a thread."""
    global _omni
    try:
        return json.dumps(connect().call(method, arguments), indent=2, ensure_ascii=False)
    except (ConnectionError, OSError):
        _omni = None  # Omni closed or restarted: the next call connects again
        raise


def error(text: str) -> types.CallToolResult:
    """A tool error the model can read and correct itself from."""
    return types.CallToolResult(is_error=True, content=[types.TextContent(type="text", text=text)])


async def list_tools(ctx, params) -> types.ListToolsResult:
    return types.ListToolsResult(tools=[tool_of(m) for m in _catalog.values()])


async def call_tool(ctx, params) -> types.CallToolResult:
    method = _catalog.get(params.name)
    if method is None:
        return error(f"unknown tool {params.name}")
    try:
        text = await asyncio.to_thread(run_call, method["name"], params.arguments or {})
        return types.CallToolResult(content=[types.TextContent(type="text", text=text)])
    except (ConnectionError, OSError) as e:
        return error(f"Omni is not reachable ({e}). Start Omni with the API on and try again.")
    except Exception as e:  # an OmniError: bad parameters, unknown track, ...
        return error(str(e))


def load_catalog() -> None:
    """Read the tool list from Omni, waiting for it to start if needed."""
    global _omni
    deadline = time.monotonic() + WAIT_FOR_OMNI_SECONDS
    while True:
        try:
            _catalog.update({m["name"].replace(".", "_"): m for m in connect().methods.values()})
            return
        except (ConnectionError, OSError) as e:
            _omni = None
            if time.monotonic() > deadline:
                sys.exit(f"Omni is not running with the API on ({e}). Start it and try again.")
            time.sleep(1)


async def main() -> None:
    server = Server(
        "omni-example",
        instructions=INSTRUCTIONS,
        on_list_tools=list_tools,
        on_call_tool=call_tool,
    )
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    load_catalog()
    asyncio.run(main())

Zarejestruj go w asystencie jak każdy serwer stdio. Dla Claude Desktop, w jego pliku konfiguracyjnym:

{
  "mcpServers": {
    "omni": {
      "command": "python",
      "args": ["C:/path/to/omni_mcp_server.py"]
    }
  }
}

Dla Claude Code: claude mcp add omni -- python C:/path/to/omni_mcp_server.py.

Test bez asystenta

SDK ma klienta, więc możesz sprawdzić serwer ze skryptu i zobaczyć dokładnie to, co zobaczyłby asystent:

import asyncio, sys
from mcp import Client
from mcp.client.stdio import StdioServerParameters

async def main():
    params = StdioServerParameters(command=sys.executable, args=["omni_mcp_server.py"])
    async with Client(params) as client:
        tools = (await client.list_tools()).tools
        print(len(tools), "tools")
        result = await client.call_tool("theory_scale", {"root": "A", "mode": "minor"})
        print(result.content[0].text[:200])

asyncio.run(main())

Uruchom go z folderu, w którym leżą omnidaw.py i omni_mcp_server.py, przy działającym Omni z włączonym API.

Co robi dobry serwer

Łączy się leniwie i łączy ponownie. Omni może wystartować po Twoim serwerze i może się zrestartować w trakcie pracy. Otwórz połączenie przy pierwszym użyciu. Jeśli wywołanie skończy się błędem połączenia, zapomnij połączenie; następne wywołanie przeczyta endpoint.json jeszcze raz (port i token zmieniają się przy każdym uruchomieniu Omni). Przykład wyżej tak robi.

Nie wysyła ponownie edycji, której odpowiedzi nie widział. Jeśli połączenie zerwie się w środku wywołania edit lub destructive, nie wiesz, czy Omni je wykonało. Powiedz modelowi, żeby najpierw sprawdził stan (tracks_list, clips_list). Odczyty zawsze można bezpiecznie powtórzyć.

Zwraca błędy jako błędy narzędzi. Komunikaty błędów Omni są pisane dla modelu i często wskazują metodę, której użyć zamiast tej. Przekaż tekst jako wynik błędu MCP (isError), a asystent zwykle poprawi się przy następnym wywołaniu. Nie zamieniaj ich na błędy protokołu.

Obsługuje długą pracę jako zadania. analysis_capture, export_render, export_stems, audio_transcribe i audio_split_stems od razu zwracają id zadania. Model wywołuje jobs_wait (do 9 sekund na wywołanie), aż zadanie się skończy. Napisz o tym w instructions serwera, tak jak robi to przykład.

Oferuje prompty. Prompty MCP to gotowe polecenia. Serwer Omni ma sześć: mix_review, loudness_check, masking_check, master_for_target, arrange_song i compose_loop. Każdy prowadzi asystenta przez pętlę, na przykład nagraj, przeanalizuj, zaproponuj, zastosuj, nagraj ponownie. To miejsce na opinie o tym, co to dobry miks; samo API pozostaje rzeczowe.

Chroni token. Nigdy nie umieszczaj go w wyniku narzędzia, prompcie ani linii logu.

Pilnuje liczby narzędzi. 102 narzędzia to dużo dla małych modeli. Jeśli Twój sobie nie radzi, wystaw podzbiór według domen (na przykład tracks, clips, theory, mixer) i dodawaj resztę na prośbę asystenta. Domena to część nazwy metody przed kropką, a każdy wpis katalogu ma ją w polu domain.

Katalog bez Omni

Dołączony serwer potrafi wypisać katalog bez działającego Omni, co przydaje się do generowania dokumentacji, typów lub testów:

omni-mcp --catalog > omni-api.json

Plik to ten sam dokument, który zwraca api.methods.

Co dodaje dołączony serwer

omni-mcp jest instalowany razem z Omni: obok omni_host.exe w Windows i wewnątrz AppImage na Linuksie, gdzie startuje jako Omni.AppImage --mcp. Oferuje te same narzędzia co przykład wyżej, a do tego kilka rzeczy, które możesz skopiować:

• Narzędzia przed startem Omni. Katalog jest w nim wbudowany, więc lista narzędzi istnieje, nawet gdy Omni nie działa. Serwer w innym języku czyta katalog z działającego Omni, jak przykład w Pythonie, albo z kopii zapisanej przez omni-mcp --catalog.
• Prompty i zasoby. Sześć promptów z góry oraz dwa zasoby: katalog API i bieżący przegląd sesji (projekt, transport i wszystkie tory).
• Ostrożne łączenie ponowne. Po restarcie Omni łączy się samo, a jeśli połączenie zerwie się w trakcie wywołania, nie powtarza go.

Jeśli piszesz w Ruście, przykład session_info.rs w paczce do pobrania pokazuje cały protokół przy użyciu samego serde_json.