feat(*): t3code-mcp connector - machines, projects, dispatch, wait, interrupt

src/t3code_mcp/config.py

@@ -0,0 +1,119 @@
+"""Machines and their project allowlists, from TOML; tokens only from env.
+
+```toml
+[machines.mac]
+url = "http://100.65.207.48:3773"
+token_env = "T3_MAC_TOKEN"
+projects = ["t3-smoke", "/Users/h/projects/openprise/beaver/*"]
+model = "claudeAgent/claude-opus-5"
+options = { effort = "high", contextWindow = "1m" }
+```
+
+`projects` entries are `fnmatch` patterns matched against a project's title
+and its workspace root. A machine without `model` uses the project default.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import os
+import tomllib
+from pathlib import Path
+from typing import Any, Self
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class ModelSpec(BaseModel):
+    """`instance/model` plus provider options, as T3's `ModelSelection`."""
+
+    model_config = ConfigDict(frozen=True)
+
+    instance: str
+    model: str
+    options: dict[str, str] = Field(default_factory=dict)
+
+    @classmethod
+    def parse(cls, text: str, options: dict[str, str] | None = None) -> Self:
+        instance, sep, model = text.partition("/")
+        if not sep or not instance or not model:
+            msg = f"model must be `instance/model`, got {text!r}"
+            raise ValueError(msg)
+        return cls(instance=instance, model=model, options=options or {})
+
+    @classmethod
+    def from_selection(cls, selection: dict[str, Any]) -> Self:
+        options = {o["id"]: str(o["value"]) for o in selection.get("options") or []}
+        return cls(
+            instance=selection["instanceId"], model=selection["model"], options=options
+        )
+
+    def selection(self) -> dict[str, Any]:
+        wire: dict[str, Any] = {"instanceId": self.instance, "model": self.model}
+        if self.options:
+            wire["options"] = [{"id": k, "value": v} for k, v in self.options.items()]
+        return wire
+
+    def __str__(self) -> str:
+        suffix = "".join(f" {k}={v}" for k, v in self.options.items())
+        return f"{self.instance}/{self.model}{suffix}"
+
+
+class Machine(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    name: str
+    url: str
+    token_env: str
+    projects: tuple[str, ...] = ()
+    model: ModelSpec | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def _fold_model(cls, data: Any) -> Any:
+        if isinstance(data, dict) and isinstance(data.get("model"), str):
+            data = {
+                **data,
+                "model": ModelSpec.parse(data["model"], data.pop("options", None)),
+            }
+        return data
+
+    @property
+    def token(self) -> str:
+        token = os.environ.get(self.token_env, "").strip()
+        if not token:
+            msg = f"machine {self.name!r}: env {self.token_env} is empty"
+            raise ValueError(msg)
+        return token
+
+    def allows(self, project: dict[str, Any]) -> bool:
+        candidates = (project.get("title", ""), project.get("workspaceRoot", ""))
+        return any(
+            fnmatch.fnmatchcase(c, pattern)
+            for pattern in self.projects
+            for c in candidates
+        )
+
+
+def load_machines(path: Path) -> dict[str, Machine]:
+    with path.open("rb") as f:
+        raw = tomllib.load(f)
+    machines = {
+        name: Machine(name=name, **fields)
+        for name, fields in raw.get("machines", {}).items()
+    }
+    if not machines:
+        msg = f"{path}: no [machines.<name>] sections"
+        raise ValueError(msg)
+    for machine in machines.values():
+        _ = machine.token
+    return machines
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="T3CODE_MCP_")
+
+    config: Path = Path("t3code.toml")
+    host: str = "0.0.0.0"
+    port: int = 8000