#!/usr/bin/env python3
"""
Chatlog MCP Server
包装现有 HTTP API 为 MCP 服务器
"""
import asyncio
import httpx
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
from typing import Any, Dict, List, Optional
import json
# HTTP API 配置
API_BASE_URL = "http://127.0.0.1:5030/api/v1"
# 创建服务器实例
server = Server("chatlog")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""返回可用的工具列表"""
return [
types.Tool(
name="list_sessions",
description="获取会话列表",
inputSchema={
"type": "object",
"properties": {
"format": {
"type": "string",
"description": "输出格式：json/text（可选）",
"default": "text"
}
}
}
),
types.Tool(
name="list_chatrooms",
description="获取群聊列表",
inputSchema={
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "搜索关键词（可选）"
},
"format": {
"type": "string",
"description": "输出格式：json/text（可选）",
"default": "text"
}
}
}
),
types.Tool(
name="list_contacts",
description="获取联系人列表",
inputSchema={
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "搜索关键词（可选）"
},
"format": {
"type": "string",
"description": "输出格式：json/text（可选）",
"default": "text"
}
}
}
),
types.Tool(
name="get_chatlog",
description="获取聊天记录",
inputSchema={
"type": "object",
"properties": {
"time": {
"type": "string",
"description": "时间范围，如 '2026-01-10' 或 '2026-01-10~2026-01-13'"
},
"talker": {
"type": "string",
"description": "聊天对象 ID 或群聊 ID"
},
"sender": {
"type": "string",
"description": "发送者 ID（可选）"
},
"keyword": {
"type": "string",
"description": "关键词搜索（可选）"
},
"format": {
"type": "string",
"description": "输出格式：json/text（可选）",
"default": "text"
}
},
"required": []
}
)
]
async def call_api(endpoint: str, params: Dict[str, Any]) -> str:
"""调用 HTTP API"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{API_BASE_URL}/{endpoint}",
params=params,
timeout=30
)
response.raise_for_status()
return response.text
except Exception as e:
return f"API 调用失败: {str(e)}"
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""处理工具调用"""
try:
if name == "list_sessions":
params = {"format": arguments.get("format", "text")}
result = await call_api("session", params)
return [types.TextContent(type="text", text=result)]
elif name == "list_chatrooms":
params = {"format": arguments.get("format", "text")}
if arguments.get("keyword"):
params["keyword"] = arguments["keyword"]
result = await call_api("chatroom", params)
return [types.TextContent(type="text", text=result)]
elif name == "list_contacts":
params = {"format": arguments.get("format", "text")}
if arguments.get("keyword"):
params["keyword"] = arguments["keyword"]
result = await call_api("contact", params)
return [types.TextContent(type="text", text=result)]
elif name == "get_chatlog":
params = {}
# 只有当参数存在时才添加到请求中
if arguments.get("time"):
params["time"] = arguments["time"]
if arguments.get("talker"):
params["talker"] = arguments["talker"]
if arguments.get("sender"):
params["sender"] = arguments["sender"]
if arguments.get("keyword"):
params["keyword"] = arguments["keyword"]
if arguments.get("format"):
params["format"] = arguments["format"]
result = await call_api("chatlog", params)
return [types.TextContent(type="text", text=result)]
else:
return [types.TextContent(type="text", text=f"未知工具: {name}")]
except Exception as e:
return [types.TextContent(type="text", text=f"错误: {str(e)}")]
async def main():
"""主函数"""
# 使用 stdio 传输启动服务器
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="chatlog",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
if __name__ == "__main__":
print("Chatlog MCP Server 启动中...")
print(f"API 基础地址: {API_BASE_URL}")
print("服务器准备就绪")
asyncio.run(main())
