mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 17:47:24 +01:00
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
file length / file-length (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
palette drift / palette (push) Waiting to run
repo-contract / repo contract (push) Waiting to run
shared gate tests / gate unit tests (push) Waiting to run
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
"""Everything a subcommand needs, injectable so tests never touch real state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from todo_hooks.client import McpAuthError, McpClient, McpError, McpUnavailable
|
|
from todo_hooks.paths import Paths
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable, Mapping
|
|
|
|
from todo_hooks.client import TaskClient
|
|
|
|
PREFIX = "todo-hooks: "
|
|
|
|
|
|
@dataclass
|
|
class Context:
|
|
"""Environment, working directory, paths, client factory and output sink."""
|
|
|
|
env: Mapping[str, str]
|
|
cwd: Path
|
|
paths: Paths
|
|
make_client: Callable[[], TaskClient]
|
|
out: Callable[[str], None]
|
|
|
|
@classmethod
|
|
def build(
|
|
cls,
|
|
env: Mapping[str, str] | None = None,
|
|
cwd: Path | None = None,
|
|
make_client: Callable[[], TaskClient] | None = None,
|
|
budget: float | None = None,
|
|
) -> Context:
|
|
"""Real defaults: ``os.environ``, the cwd, ``~`` paths, a live client."""
|
|
use_env = dict(os.environ) if env is None else dict(env)
|
|
paths = Paths.default(use_env)
|
|
|
|
def live() -> TaskClient:
|
|
return McpClient.from_env(use_env, paths.token_file, budget=budget)
|
|
|
|
return cls(
|
|
env=use_env,
|
|
cwd=cwd or Path.cwd(),
|
|
paths=paths,
|
|
make_client=make_client or live,
|
|
out=print,
|
|
)
|
|
|
|
def say(self, message: str) -> None:
|
|
self.out(PREFIX + message)
|
|
|
|
|
|
def describe(err: McpError) -> str:
|
|
"""Server-side failures read as 'unavailable'; tool errors verbatim."""
|
|
if isinstance(err, McpUnavailable | McpAuthError):
|
|
return f"todo server unavailable ({err})"
|
|
return str(err)
|