surge.py•1.37 kB
import logging
from datetime import datetime, timedelta
from lib.supabase import supabase
from mcp.server.fastmcp import FastMCP
# 配置日志记录
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 初始化 MCP Server
mcp = FastMCP("surge", port=8000,host="0.0.0.0")
def get_resources_list() -> list[dict]:
response = supabase.table("ai_surge_resource_list").select("name,channel").execute()
return response.data
def get_recent_news_data() -> list[dict]:
response = supabase.table("ai_surge_news_list").select("title,desc,thumbnails,ai_summary,video_url,published_at").gte("published_at", (datetime.now() - timedelta(days=7)).isoformat()).order("published_at",desc=True).execute()
return response.data
@mcp.tool()
async def get_all_resources():
"""
Get all resources list
Returns:
list[dict]: A list of resources
"""
return get_resources_list()
@mcp.tool()
async def get_recent_news():
"""
Get recent news
Returns:
list[dict]: A list of news
"""
return get_recent_news_data()
if __name__ == "__main__":
try:
logger.info("Surge MCP Server starting on host 0.0.0.0, port 8888...")
mcp.run(transport='sse')
except Exception as e:
logger.error(f"Failed to start server: {e}")
raise