betterbot/tools/meta/__init__.py

"""Credential meta-tools (T039-T041).

Provides tools for users to inspect their service access and credentials.
"""

from __future__ import annotations

from typing import Any

from tool_registry import ToolSet, openai_tools_to_copilot

META_SYSTEM_PROMPT = """\
You can help users check their service access and available integrations. \
When a user asks about their accounts, services, or credentials, use the \
list_my_services and get_my_credentials tools. \
Never reveal passwords unless the system configuration explicitly allows it.
"""

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_my_services",
            "description": (
                "List all services available to the user, showing which are active "
                "(have credentials), which are available for setup, and their capabilities."
            ),
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_my_credentials",
            "description": (
                "Get the user's credentials for a specific service. Returns the username "
                "and service URL. Password is only shown if system configuration allows it."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "service": {
                        "type": "string",
                        "description": "The service name (e.g. 'vikunja', 'karakeep')",
                    },
                },
                "required": ["service"],
            },
        },
    },
]


def _build_meta_tools(user_context: dict[str, Any]) -> list:
    """Factory that creates Copilot SDK meta-tools."""
    user = user_context.get("_user")

    async def dispatcher(name: str, arguments: dict, **_kw: Any) -> str:
        if name == "list_my_services":
            return _handle_list_my_services(user)
        if name == "get_my_credentials":
            return _handle_get_my_credentials(user, arguments.get("service", ""))
        return f"Unknown meta-tool: {name}"

    return openai_tools_to_copilot(schemas=TOOLS, dispatcher=dispatcher)


def _handle_list_my_services(user: Any) -> str:
    if user is None:
        return "Unable to determine your identity."

    from tool_pipeline import get_provisionable_services

    services = get_provisionable_services(user)
    if not services:
        return "No services are currently configured."

    lines = []
    for s in services:
        status = s["status"].upper()
        caps = s["capabilities"] or "general"
        lines.append(f"- **{s['service']}** ({caps}): {status}")
    return "Your services:\n" + "\n".join(lines)


def _handle_get_my_credentials(user: Any, service: str) -> str:
    if user is None:
        return "Unable to determine your identity."
    if not service:
        return "Please specify which service you want credentials for."

    from config import settings
    from user_store import get_store

    store = get_store()
    cred = store.get_credential(user.id, service)
    if cred is None:
        return f"You don't have credentials for {service}. Say 'set up {service}' to get started."

    parts = [f"**Service**: {service}"]
    if cred.service_username:
        parts.append(f"**Username**: {cred.service_username}")

    # Service URL from settings
    url_attr = f"{service.upper()}_API_URL"
    url = getattr(settings, url_attr, "")
    if url:
        # Strip /api/v1 suffix for display
        display_url = url.rstrip("/")
        for suffix in ("/api/v1", "/api"):
            if display_url.endswith(suffix):
                display_url = display_url[: -len(suffix)]
                break
        parts.append(f"**URL**: {display_url}")

    if cred.expires_at:
        parts.append(f"**Expires**: {cred.expires_at}")

    if settings.ALLOW_CREDENTIAL_REVEAL_IN_CHAT:
        from user_store import decrypt

        try:
            token = decrypt(cred.encrypted_token)
            parts.append(f"**API Token**: `{token}`")
            parts.append("⚠️ Be careful — this token grants full access to your account.")
        except Exception:
            parts.append("**API Token**: (decryption failed)")
    else:
        parts.append("**Password/Token**: Stored securely. Access the service directly at the URL above.")

    return "\n".join(parts)


meta_toolset = ToolSet(
    name="meta",
    description="Service access and credential information",
    capability="account_management",
    system_prompt_fragment=META_SYSTEM_PROMPT,
    build_tools=_build_meta_tools,
    required_keys=[],  # Available to all authenticated users
)