File size: 2,373 Bytes
1a9e2c2 | 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 | # -*- coding: utf-8 -*-
"""MCP Tools - Grok AI 对话工具"""
import json
from typing import Optional
from app.services.grok.client import GrokClient
from app.core.logger import logger
from app.core.exception import GrokApiException
async def ask_grok_impl(
query: str,
model: str = "grok-3-fast",
system_prompt: Optional[str] = None
) -> str:
"""
内部实现: 调用Grok API并收集完整响应
Args:
query: 用户问题
model: 模型名称
system_prompt: 系统提示词
Returns:
str: 完整的Grok响应内容
"""
try:
# 构建消息列表
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": query})
# 构建请求
request_data = {
"model": model,
"messages": messages,
"stream": True
}
logger.info(f"[MCP] ask_grok 调用, 模型: {model}")
# 调用Grok客户端(流式)
response_iterator = await GrokClient.openai_to_grok(request_data)
# 收集所有流式响应块
content_parts = []
async for chunk in response_iterator:
if isinstance(chunk, bytes):
chunk = chunk.decode('utf-8')
# 解析SSE格式
if chunk.startswith("data: "):
data_str = chunk[6:].strip()
if data_str == "[DONE]":
break
try:
data = json.loads(data_str)
choices = data.get("choices", [])
if choices:
delta = choices[0].get("delta", {})
if content := delta.get("content"):
content_parts.append(content)
except json.JSONDecodeError:
continue
result = "".join(content_parts)
logger.info(f"[MCP] ask_grok 完成, 响应长度: {len(result)}")
return result
except GrokApiException as e:
logger.error(f"[MCP] Grok API错误: {str(e)}")
raise Exception(f"Grok API调用失败: {str(e)}")
except Exception as e:
logger.error(f"[MCP] ask_grok异常: {str(e)}", exc_info=True)
raise Exception(f"处理请求时出错: {str(e)}")
|