root
9c53f58165
All checks were successful
构建并部署 AI Agent 服务 / deploy (push) Successful in 5m38s
- 添加MCP Manager统一入口管理 - 实现Contact/Dictionary/News三个适配器 - 三层降级策略：MCP -> Database -> Mock - 保持原有api_client向后兼容 - 添加完整文档和测试
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""
|
||
词典API调用工具（使用MCP统一接口）
|
||
"""
|
||
from typing import Dict, Any, Optional
|
||
from dataclasses import dataclass
|
||
|
||
from ...mcp.mcp_manager import mcp_manager
|
||
from ...mcp.adapters import DictionaryAdapter
|
||
|
||
|
||
@dataclass
|
||
class DictionaryAPIClient:
|
||
"""
|
||
词典API客户端 - 使用MCP统一接口
|
||
|
||
保持向后兼容，内部使用MCP适配器
|
||
"""
|
||
|
||
# 保留配置字段用于向后兼容
|
||
youdao_api_key: Optional[str] = None
|
||
youdao_api_secret: Optional[str] = None
|
||
word_repository: Optional[Any] = None
|
||
|
||
def __post_init__(self):
|
||
"""初始化后设置MCP"""
|
||
import asyncio
|
||
try:
|
||
asyncio.create_task(self._init_mcp())
|
||
except RuntimeError:
|
||
pass
|
||
|
||
async def _init_mcp(self):
|
||
"""初始化MCP系统"""
|
||
if not mcp_manager.get_adapter("dictionary"):
|
||
mcp_manager.register_adapter(
|
||
DictionaryAdapter(word_repo=self.word_repository)
|
||
)
|
||
await mcp_manager.initialize()
|
||
|
||
async def query_word(
|
||
self,
|
||
user_id: str = "default",
|
||
word: str = "",
|
||
use_cache: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""查询单词（统一入口）"""
|
||
await self._init_mcp()
|
||
result = await mcp_manager.execute(
|
||
"dictionary", "query_word",
|
||
user_id=user_id, word=word, use_cache=use_cache
|
||
)
|
||
if result.success:
|
||
return result.data
|
||
return self.query_word_mock(word)
|
||
|
||
def query_word_mock(self, word: str) -> Dict[str, Any]:
|
||
"""模拟查询（保留用于向后兼容）"""
|
||
return {
|
||
"word": word,
|
||
"phonetic": "",
|
||
"part_of_speech": "n.",
|
||
"definitions": [f"{word} 的释义1", f"{word} 的释义2"],
|
||
"examples": [f"This is an example sentence with '{word}'."]
|
||
}
|
||
|
||
def translate_mock(self, text: str, from_lang: str = "auto", to_lang: str = "zh") -> Dict[str, Any]:
|
||
"""模拟翻译（保留用于向后兼容）"""
|
||
return {
|
||
"translated_text": f"【翻译】{text}",
|
||
"confidence": 0.95
|
||
}
|
||
|
||
def extract_terms_mock(self, text: str) -> list:
|
||
"""模拟术语提取（保留用于向后兼容）"""
|
||
return [
|
||
{"term": "AI", "type": "技术术语", "definition": "人工智能", "confidence": 0.95},
|
||
{"term": "大模型", "type": "技术术语", "definition": "大语言模型", "confidence": 0.92}
|
||
]
|
||
|
||
|
||
# 全局单例（保持向后兼容）
|
||
dictionary_api = DictionaryAPIClient()
|