File size: 3,398 Bytes
bdcdaf4 d0c18f0 bdcdaf4 cc826a1 bdcdaf4 cc826a1 bdcdaf4 cc826a1 bdcdaf4 d0c18f0 cc826a1 d0c18f0 cc826a1 d0c18f0 cc826a1 d0c18f0 bdcdaf4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """
文件缓存插件 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) |