123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
"""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
|
|
state: Path = Path("t3code-mcp.json")
|
|
hook_url: str | None = None
|
|
gateway_token: str | None = None
|