response_formatter.py•10.8 kB
"""
响应格式化工具
提供统一的响应格式化功能，支持JSON和Markdown格式。
"""
import json
from typing import Dict, Any, List, Optional, Union
from datetime import datetime
from ..models import (
ResponseMetadata, create_error_response, generate_request_id
)
class ResponseFormatter:
"""响应格式化器"""
@staticmethod
def format_success_response(
data: Any,
response_format: str = "json",
metadata: Optional[ResponseMetadata] = None,
include_metadata: bool = True
) -> Dict[str, Any]:
"""
格式化成功响应
Args:
data: 响应数据
response_format: 响应格式 (json/markdown)
metadata: 响应元数据
include_metadata: 是否包含元数据
Returns:
格式化后的响应
"""
# 创建元数据（如果未提供）
if metadata is None and include_metadata:
metadata = ResponseMetadata(
request_id=generate_request_id(),
timestamp=datetime.now().isoformat(),
status="success"
)
# 构建响应
response = {
"status": "success",
"data": data
}
# 添加元数据（如果需要）
if include_metadata and metadata:
response["metadata"] = metadata.to_dict()
# 根据格式返回
if response_format.lower() == "markdown":
return ResponseFormatter._to_markdown(response)
else:
return response
@staticmethod
def format_error_response(
error_code: str,
message: str,
details: Optional[Dict[str, Any]] = None,
response_format: str = "json",
metadata: Optional[ResponseMetadata] = None,
include_metadata: bool = True
) -> Dict[str, Any]:
"""
格式化错误响应
Args:
error_code: 错误代码
message: 错误消息
details: 错误详情
response_format: 响应格式 (json/markdown)
metadata: 响应元数据
include_metadata: 是否包含元数据
Returns:
格式化后的错误响应
"""
# 创建错误响应
error_response = create_error_response(
error_code=error_code,
message=message,
details=details
)
# 创建元数据（如果未提供）
if metadata is None and include_metadata:
metadata = ResponseMetadata(
request_id=generate_request_id(),
timestamp=datetime.now().isoformat(),
status="error"
)
# 添加元数据（如果需要）
if include_metadata and metadata:
error_response["metadata"] = metadata.to_dict()
# 根据格式返回
if response_format.lower() == "markdown":
return ResponseFormatter._to_markdown(error_response)
else:
return error_response
@staticmethod
def format_article_response(
article_data: Dict[str, Any],
response_format: str = "json",
concise: bool = False,
include_metadata: bool = True
) -> Dict[str, Any]:
"""
格式化文章响应
Args:
article_data: 文章数据
response_format: 响应格式 (json/markdown)
concise: 是否返回简洁格式
include_metadata: 是否包含元数据
Returns:
格式化后的文章响应
"""
# 处理简洁格式
if concise:
concise_data = {
"title": article_data.get("title", ""),
"author": article_data.get("author", ""),
"publish_time": article_data.get("publish_time", ""),
"summary": article_data.get("summary", ""),
"url": article_data.get("url", "")
}
article_data = concise_data
return ResponseFormatter.format_success_response(
data=article_data,
response_format=response_format,
include_metadata=include_metadata
)
@staticmethod
def format_search_response(
search_results: List[Dict[str, Any]],
response_format: str = "json",
concise: bool = False,
include_metadata: bool = True
) -> Dict[str, Any]:
"""
格式化搜索响应
Args:
search_results: 搜索结果列表
response_format: 响应格式 (json/markdown)
concise: 是否返回简洁格式
include_metadata: 是否包含元数据
Returns:
格式化后的搜索响应
"""
# 处理简洁格式
if concise:
concise_results = []
for result in search_results:
concise_result = {
"title": result.get("title", ""),
"author": result.get("author", ""),
"publish_time": result.get("publish_time", ""),
"summary": result.get("summary", ""),
"url": result.get("url", "")
}
concise_results.append(concise_result)
search_results = concise_results
return ResponseFormatter.format_success_response(
data={
"results": search_results,
"count": len(search_results)
},
response_format=response_format,
include_metadata=include_metadata
)
@staticmethod
def format_summary_response(
summary_data: Dict[str, Any],
response_format: str = "json",
include_metadata: bool = True
) -> Dict[str, Any]:
"""
格式化摘要响应
Args:
summary_data: 摘要数据
response_format: 响应格式 (json/markdown)
include_metadata: 是否包含元数据
Returns:
格式化后的摘要响应
"""
return ResponseFormatter.format_success_response(
data=summary_data,
response_format=response_format,
include_metadata=include_metadata
)
@staticmethod
def _to_markdown(data: Dict[str, Any]) -> Dict[str, Any]:
"""
将响应数据转换为Markdown格式
Args:
data: 响应数据
Returns:
包含Markdown内容的响应
"""
# 创建Markdown内容
markdown_content = ResponseFormatter._dict_to_markdown(data)
# 返回Markdown响应
return {
"content_type": "text/markdown",
"content": markdown_content
}
@staticmethod
def _dict_to_markdown(
data: Dict[str, Any],
level: int = 0
) -> str:
"""
将字典转换为Markdown
Args:
data: 字典数据
level: 嵌套级别
Returns:
Markdown字符串
"""
markdown = []
for key, value in data.items():
# 跳过元数据（如果需要）
if key == "metadata" and level == 0:
continue
# 处理标题
if level == 0:
markdown.append(f"## {key.replace('_', ' ').title()}")
else:
markdown.append(f"### {key.replace('_', ' ').title()}")
# 处理值
if isinstance(value, dict):
markdown.append(ResponseFormatter._dict_to_markdown(value, level + 1))
elif isinstance(value, list):
for item in value:
if isinstance(item, dict):
markdown.append(ResponseFormatter._dict_to_markdown(item, level + 1))
else:
markdown.append(f"- {item}")
elif isinstance(value, str) and len(value) > 200:
# 长文本处理
markdown.append(f"```\n{value}\n```")
else:
markdown.append(f"{value}")
markdown.append("") # 空行
return "\n".join(markdown)
def format_mcp_tool_response(
tool_name: str,
result: Any,
response_format: str = "json",
concise: bool = False,
error: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
格式化MCP工具响应
Args:
tool_name: 工具名称
result: 工具结果
response_format: 响应格式 (json/markdown)
concise: 是否返回简洁格式
error: 错误信息（如果有）
Returns:
MCP工具响应列表
"""
if error:
# 错误响应
error_response = ResponseFormatter.format_error_response(
error_code=error.get("code", "UNKNOWN_ERROR"),
message=error.get("message", "Unknown error"),
details=error.get("details"),
response_format=response_format
)
if response_format.lower() == "markdown":
return [{"type": "text", "text": error_response["content"]}]
else:
return [{"type": "text", "text": json.dumps(error_response, ensure_ascii=False, indent=2)}]
else:
# 成功响应
if tool_name == "fetch_article":
response = ResponseFormatter.format_article_response(
article_data=result,
response_format=response_format,
concise=concise
)
elif tool_name == "search_articles":
response = ResponseFormatter.format_search_response(
search_results=result.get("results", []),
response_format=response_format,
concise=concise
)
elif tool_name == "extract_article_summary":
response = ResponseFormatter.format_summary_response(
summary_data=result,
response_format=response_format
)
else:
# 通用响应
response = ResponseFormatter.format_success_response(
data=result,
response_format=response_format
)
if response_format.lower() == "markdown":
return [{"type": "text", "text": response["content"]}]
else:
return [{"type": "text", "text": json.dumps(response, ensure_ascii=False, indent=2)}]