beaver-agent/config.py

import os
from datetime import date
from pathlib import Path

from beaver_gateway.agents.base import ExposedMcp
from beaver_gateway.agents.claude import ClaudeAgent, ClaudeCodeOptions
from beaver_gateway.agents.raycast import RaycastAgent, RemoteTool, UserPreferences
from beaver_gateway.core.registry import Gateway
from beaver_gateway.core.turn_record import TurnRecord, slugify
from beaver_gateway.frontends.admin import AdminFrontend
from beaver_gateway.frontends.anthropic import AnthropicMessagesFrontend
from beaver_gateway.frontends.markdown import MarkdownFrontend
from beaver_gateway.frontends.mcp_server import McpServerFrontend
from beaver_gateway.mcp.types import HttpMcp, McpServer


VAULT = Path("/vault")
CHATS_DIR = VAULT / "💬 чаты"


def chat_log_path(record: TurnRecord, vault: Path) -> Path:
    today = date.today()
    topic = slugify(record.first_user_text, maxlen=60)
    rel = (
        CHATS_DIR.relative_to(vault)
        / f"{today:%Y-%m}"
        / f"{today:%Y-%m-%d} - {topic}.md"
    )
    return vault / rel


def _calendar_mcps() -> list[HttpMcp]:
    raw = os.environ.get("CALENDAR_MCPS", "").strip()
    servers: list[HttpMcp] = []
    for entry in raw.split(","):
        entry = entry.strip()
        if not entry:
            continue
        name, sep, url = entry.partition("=")
        if not sep or not url.strip():
            raise ValueError(f"CALENDAR_MCPS entry must be `name=url`, got: {entry!r}")
        servers.append(McpServer.http(name=f"calendar-{name.strip()}", url=url.strip()))
    return servers


calendar_mcps = _calendar_mcps()
calendar_exposed = tuple(ExposedMcp(name=m.name) for m in calendar_mcps)

mcps = [
    McpServer.stdio(
        name="obsidian-fs",
        command=[
            "bunx",
            "-y",
            "@modelcontextprotocol/server-filesystem",
            "/vault",
        ],
        lenient=True,
    ),
    McpServer.stdio(
        name="firefly",
        command=[
            "bunx",
            "-y",
            "@firefly-iii-mcp/local",
            "--pat",
            os.environ["FIREFLY_PAT"],
            "--baseUrl",
            os.environ["FIREFLY_BASE_URL"],
            "--preset",
            "default",
        ],
        lenient=True,
    ),
    McpServer.http(name="telegram", url=os.environ["BEAVERGRAM_MCP"]),
    *calendar_mcps,
]


CBO_PROMPT = (Path(__file__).parent / "prompt.md").read_text(encoding="utf-8")


UserPrefsRu = lambda: UserPreferences(  # noqa: E731
    locale="ru-RU",
    timezone="Europe/Warsaw",
    current_date=date.today().isoformat(),  # noqa: DTZ011
)


def claude(name: str, model: str, effort: str | None = None) -> ClaudeAgent:
    return ClaudeAgent(
        name=name,
        model=model,
        system_prompt=CBO_PROMPT,
        cwd=VAULT,
        options=ClaudeCodeOptions(
            effort=effort,
            extra_args=("--remote-control",),
            disallowed_tools=("AskUserQuestion", "ExitPlanMode", "EnterPlanMode"),
        ),
        expose_mcps=(
            ExposedMcp(name="firefly"),
            ExposedMcp(name="telegram"),
            *calendar_exposed,
        ),
    )


def raycast(name: str, model: str, reasoning_effort: str | None = None) -> RaycastAgent:
    return RaycastAgent(
        name=name,
        model=model,
        system_prompt=CBO_PROMPT,
        reasoning_effort=reasoning_effort,
        available_native_tools=(RemoteTool.WEB_SEARCH, RemoteTool.READ_PAGE),
        user_preferences=UserPrefsRu,
        expose_mcps=(
            ExposedMcp(name="obsidian-fs"),
            ExposedMcp(name="firefly"),
            ExposedMcp(name="telegram"),
            *calendar_exposed,
        ),
    )


agents = [
    claude("beaver-opus-high", "claude-opus-4-8", effort="high"),
    raycast("beaver-gemini-pro-high", "google-gemini-3.1-pro", reasoning_effort="high"),
    claude("beaver-opus-medium", "claude-opus-4-8", effort="medium"),
    claude("beaver-opus-xhigh", "claude-opus-4-8", effort="xhigh"),
    raycast("beaver-gemini-pro-low", "google-gemini-3.1-pro", reasoning_effort="low"),
    raycast(
        "beaver-gemini-flash-high", "google-gemini-3.5-flash", reasoning_effort="high"
    ),
    raycast(
        "beaver-gemini-flash-low", "google-gemini-3.5-flash", reasoning_effort="low"
    ),
]

PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "").rstrip("/")


def _public(suffix: str) -> str | None:
    return f"{PUBLIC_BASE_URL}{suffix}" if PUBLIC_BASE_URL else None


frontends = [
    AnthropicMessagesFrontend(
        host="0.0.0.0", port=62990, public_base_url=_public("/anthropic")
    ),
    McpServerFrontend(host="0.0.0.0", port=62991, public_base_url=_public("/mcp")),
    AdminFrontend(host="0.0.0.0", port=62992, public_base_url=_public("/admin")),
    MarkdownFrontend(
        host="0.0.0.0",
        port=62993,
        vault_path=CHATS_DIR,
        default_agent="research",
        log_all_chats=True,
        log_path=chat_log_path,
        public_base_url=_public("/md"),
    ),
]


gateway = Gateway(agents=agents, mcps=mcps, frontends=frontends)