plugins/cache/mcp.py

"""
文件缓存插件 MCP工具定义
"""

from app.mcp.decorators import mcp_tool
from pydantic import BaseModel, Field

from .core import do_set, do_get, do_delete, do_set_text, do_get_text, do_set_from_url


class SetInput(BaseModel):
    """设置缓存输入参数"""
    key: str = Field(description="缓存键名")
    value_base64: str = Field(description="缓存值的Base64编码")
    ttl_seconds: int = Field(default=3600, ge=60, le=86400, description="过期时间（秒）")


class GetInput(BaseModel):
    """获取缓存输入参数"""
    key: str = Field(description="缓存键名")


class DeleteInput(BaseModel):
    """删除缓存输入参数"""
    key: str = Field(description="缓存键名")


class CacheOutput(BaseModel):
    """缓存操作输出结果"""
    success: bool = Field(description="操作是否成功")
    value: str | None = Field(default=None, description="缓存值（仅get操作）")
    message: str = Field(description="结果说明")


@mcp_tool(name="cache-set", title="设置缓存", description="设置缓存值，支持TTL")
async def cache_set(params: SetInput) -> CacheOutput:
    result = await do_set(params.key, params.value_base64, params.ttl_seconds)
    return CacheOutput(**result)


@mcp_tool(name="cache-get", title="获取缓存", description="获取缓存值")
async def cache_get(params: GetInput) -> CacheOutput:
    result = await do_get(params.key)
    return CacheOutput(**result)


@mcp_tool(name="cache-delete", title="删除缓存", description="删除缓存值")
async def cache_delete(params: DeleteInput) -> CacheOutput:
    result = await do_delete(params.key)
    return CacheOutput(**result)


# === 新增文本缓存工具 ===


class SetTextInput(BaseModel):
    """设置文本缓存输入参数"""
    key: str = Field(description="缓存键名")
    text_content: str = Field(description="文本内容（UTF-8）")
    ttl_seconds: int = Field(default=3600, ge=60, le=86400, description="过期时间（秒）")


class SetFromUrlInput(BaseModel):
    """从URL设置缓存输入参数"""
    key: str = Field(description="缓存键名")
    url: str = Field(description="文件URL地址")
    ttl_seconds: int = Field(default=3600, ge=60, le=86400, description="过期时间（秒）")


class TextCacheOutput(BaseModel):
    """文本缓存操作输出结果"""
    success: bool = Field(description="操作是否成功")
    text: str | None = Field(default=None, description="文本内容（仅get_text操作）")
    message: str = Field(description="结果说明")


@mcp_tool(name="cache-text", title="设置文本缓存", description="直接缓存文本内容，自动编码")
async def cache_set_text(params: SetTextInput) -> TextCacheOutput:
    result = await do_set_text(params.key, params.text_content, params.ttl_seconds)
    return TextCacheOutput(**result)


@mcp_tool(name="cache-readtext", title="获取文本缓存", description="获取文本缓存，自动解码为UTF-8")
async def cache_get_text(params: GetInput) -> TextCacheOutput:
    result = await do_get_text(params.key)
    return TextCacheOutput(**result)


@mcp_tool(name="cache-url", title="URL缓存", description="从URL下载文件并缓存")
async def cache_set_from_url(params: SetFromUrlInput) -> CacheOutput:
    result = await do_set_from_url(params.key, params.url, params.ttl_seconds)
    return CacheOutput(**result)