外观
06.企业级MCP集成方案与应用实践
来源：语雀原文 | 整理日期：2026-09-22
示例代码
MCP 架构组件介绍
MCP 核心概念
Model Context Protocol (MCP) 是由 Anthropic 于 2024年11月推出的开放标准协议，用于标准化 AI 助手与数据源之间的连接方式。MCP 协议解决了 AI 应用访问外部系统数据时的标准化问题，类似于为 AI 应用提供了一个"USB-C"接口。
核心特性：
- 开放标准：基于 JSON-RPC 的通信协议
- 双向连接：支持客户端和服务器之间的安全双向通信
- 标准化接口：为不同数据源提供统一的访问接口
- 企业级安全：内置安全机制和权限控制
MCP 协议定义了清晰的架构模式：
客户端 (MCP Client)：
- 发起连接和请求的应用程序
- 通常是 AI 应用或 LLM 集成系统
- 负责协议协商和会话管理
服务器 (MCP Server)：
- 提供数据和功能的服务提供方
- 暴露资源、工具和提示词
- 处理客户端请求并返回响应
传输层：
- 支持多种传输方式：stdio、SSE、Streamable HTTP
- 处理消息序列化和网络通信
- 提供连接管理和错误恢复
MCP 协议原语
MCP 定义了五个核心原语：
- Resources（资源）：只读数据访问接口，类似 GET 端点
- Tools（工具）：可执行功能接口，类似 POST 端点
- Prompts（提示词）：可重用的 LLM 交互模板
- Roots（根）：客户端数据源声明
- Sampling（采样）：LLM 推理请求处理
版本管理机制
MCP 使用基于日期的版本标识符：
- 格式：
YYYY-MM-DD - 当前版本：
2025-06-18 - 版本状态：Draft、Current、Final
最佳实践原则
协议设计最佳实践
- 版本兼容性评估：选择 AI 提供商时评估 MCP 兼容性
- 抽象层设计：为多 AI 系统设计抽象层和备用机制
- 标准传输协议：优先使用官方推荐的传输方式
安全最佳实践
- 权限最小化原则：仅授予必要的访问权限
- 连接加密：使用 TLS 加密传输通道
- 输入验证：严格验证所有输入数据
- 访问控制：实施细粒度的访问控制机制
python
# MCP 安全配置示例
mcp_config = {
"security": {
"require_tls": True,
"max_connections": 100,
"timeout_seconds": 30,
"allowed_origins": ["https://trusted-domain.com"]
},
"access_control": {
"enable_rbac": True,
"default_permissions": ["read"],
"admin_users": ["admin@company.com"]
}
}MCP服务器SSE开发实践
环境准备
python版本：3.12.4
plain
# 安装依赖
mcp>=1.2.0
mcp[cli]
fastmcp>=2.0.0
OpenAI
marvin
psycopg2基础服务器架构
服务器初始化
python
from fastmcp import FastMCP
mcp = FastMCP(name="base_mcp_server")
if __name__ == "__main__":
mcp.run(transport='sse', host="127.0.0.1", port=8001)资源管理
python
# 添加一个动态问候资源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""获取个性化问候"""
return f"Hello, {name}!"
@mcp.resource("db://province_info/data/{province_name}")
async def get_province_data(province_name: str) -> str:
"""中国省份（含直辖市）信息
参数:
province_name: 中国省份名称或直辖市，如广东省，北京市
返回:
以 json 格式返回某省份信息
"""
sql = '''
select region_code,
province_name,
capital_city,
area,
population,
abbreviation
from public.chinese_provinces
where province_name = '{province_name}'
'''.format(province_name=province_name)
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql)
rows = cur.fetchall()
return json.dumps(list(rows), default=str)工具实现
python
@mcp.tool()
def add(a: float, b: float) -> float:
"""加法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之和
"""
return a + b
@mcp.tool()
def subtract(a: float, b: float) -> float:
"""减法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之差 (a - b)
"""
return a - b
@mcp.tool()
def multiply(a: float, b: float) -> float:
"""乘法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之积
"""
return a * b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""除法运算
参数:
a: 被除数
b: 除数
返回:
两数之商 (a / b)
异常:
ValueError: 当除数为零时
"""
if b == 0:
raise ValueError("除数不能为零")
return a / b完整服务器实现
python
from fastmcp import FastMCP
mcp = FastMCP(name="base_mcp_server")
@mcp.tool()
def add(a: float, b: float) -> float:
"""加法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之和
"""
return a + b
@mcp.tool()
def subtract(a: float, b: float) -> float:
"""减法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之差 (a - b)
"""
return a - b
@mcp.tool()
def multiply(a: float, b: float) -> float:
"""乘法运算
参数:
a: 第一个数字
b: 第二个数字
返回:
两数之积
"""
return a * b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""除法运算
参数:
a: 被除数
b: 除数
返回:
两数之商 (a / b)
异常:
ValueError: 当除数为零时
"""
if b == 0:
raise ValueError("除数不能为零")
return a / b
if __name__ == "__main__":
mcp.run(transport='sse', host="127.0.0.1", port=8001)最佳实践要点
- 错误处理：实现完善的异常处理和错误恢复机制
- 性能优化：使用缓存和连接池优化性能
- 监控日志：集成日志记录和监控指标
- 测试覆盖：编写全面的单元测试和集成测试
MCP 客户端整合大模型开发实践
客户端架构设计
基础客户端类
python
import asyncio
import json
import logging
import os
import marvin
from fastmcp import Client
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
from marvin import settings
from openai import OpenAI, AsyncOpenAI
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.deepseek import DeepSeekProvider
settings.enable_default_print_handler = False
class LLMClient:
"""大语言模型客户端，负责与LLM API进行通信"""
def __init__(self, model_name: str, url: str, api_key: str) -> None:
"""
初始化LLM客户端
Args:
model_name: 模型名称
url: API基础URL
api_key: API密钥
"""
self.model_name: str = model_name
self.url: str = url
self.client = OpenAI(api_key=api_key, base_url=url)
def get_response(self, messages: list[dict[str, str]]) -> str:
"""
向LLM发送消息并获取响应
Args:
messages: 消息列表，每条消息包含role和content
Returns:
LLM的响应文本
"""
response = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
stream=False
)
return response.choices[0].message.content
class ChatSession:
"""聊天会话管理器，处理用户输入、LLM响应和资源访问"""
def __init__(self, llm_client: LLMClient, mcp_client: Client) -> None:
"""
初始化聊天会话
Args:
llm_client: LLM客户端实例
mcp_client: MCP客户端实例，用于访问资源
"""
self.mcp_client: Client = mcp_client
self.llm_client: LLMClient = llm_client
async def process_llm_response(self, llm_response: str) -> str:
"""
处理LLM的响应，解析资源URI调用或工具调用并执行
Args:
llm_response: LLM返回的响应文本
Returns:
处理后的响应文本，包含资源数据或工具执行结果
"""
try:
# 检查是否为资源URI调用（以db://开头）
if llm_response.strip().startswith('db://'):
uri = llm_response.strip()
try:
# 执行资源读取
resource_data = await self.mcp_client.read_resource(uri=uri)
return f"Resource data: {resource_data}"
except Exception as e:
error_msg = f"Error reading resource: {str(e)}"
logging.error(error_msg)
return error_msg
else:
# 尝试解析为JSON工具调用（保留兼容性）
if llm_response.startswith('```json'):
llm_response = llm_response.strip('```json').strip('```').strip()
try:
tool_call = json.loads(llm_response)
if "tool" in tool_call and "arguments" in tool_call:
# 检查工具是否可用
available_tools = await self.mcp_client.list_tools()
if any(tool.name == tool_call["tool"] for tool in available_tools):
try:
# 执行工具调用
tool_result = await self.mcp_client.call_tool(
tool_call["tool"], tool_call["arguments"]
)
return f"Tool execution result: {tool_result}"
except Exception as e:
error_msg = f"Error executing tool: {str(e)}"
logging.error(error_msg)
return error_msg
return f"Tool not found: {tool_call['tool']}"
except json.JSONDecodeError:
pass
# 如果不是JSON格式或工具调用，直接返回原始响应
return llm_response
except Exception as e:
error_msg = f"Error processing LLM response: {str(e)}"
logging.error(error_msg)
return error_msg
async def start(self, system_message: str) -> None:
"""
启动聊天会话的主循环
Args:
system_message: 系统提示消息，指导LLM的行为
"""
messages = [{"role": "system", "content": system_message}]
while True:
try:
# 获取用户输入
user_input = input("用户: ").strip().lower()
if user_input in ["quit", "exit", "退出"]:
print('AI助手已退出')
break
messages.append({"role": "user", "content": user_input})
# 获取LLM的初始响应
llm_response = self.llm_client.get_response(messages)
print("助手: ", llm_response)
# 处理可能的资源调用或工具调用
result = await self.process_llm_response(llm_response)
# 如果处理结果与原始响应不同，说明执行了资源调用或工具调用，需要进一步处理
while result != llm_response:
messages.append({"role": "assistant", "content": llm_response})
messages.append({"role": "system", "content": result})
# 将资源数据或工具执行结果发送回LLM获取新响应
llm_response = self.llm_client.get_response(messages)
result = await self.process_llm_response(llm_response)
print("助手: ", llm_response)
messages.append({"role": "assistant", "content": llm_response})
except KeyboardInterrupt:
print('AI助手已退出')
break
# 获取DeepSeek API密钥
api_key = os.getenv('DEEPSEEK_API_KEY')
base_url = os.getenv('DEEPSEEK_BASE_URL')
client = AsyncOpenAI(max_retries=3,base_url=base_url,api_key=api_key)
# 初始化DeepSeek模型
model = OpenAIModel(
model_name='deepseek-v3-250324',
provider=DeepSeekProvider(openai_client=client),
)
# 创建Marvin智能助手
agent = marvin.Agent(
model=model,
name="智能助手"
)
async def sampling_func(
messages: list[SamplingMessage],
params: SamplingParams,
ctx: RequestContext,
) -> str:
"""
采样函数，用于处理消息并获取LLM响应
Args:
messages: 消息列表
params: 采样参数
ctx: 请求上下文
Returns:
LLM的响应文本
"""
return await marvin.say_async(
message=[m.content.text for m in messages],
instructions=params.systemPrompt,
agent=agent
)
async def main():
"""主函数，初始化客户端并启动聊天会话"""
async with Client("http://127.0.0.1:8002/sse", sampling_handler=sampling_func,timeout=30000000) as mcp_client:
# 初始化LLM客户端，使用通义千问模型
llm_client = LLMClient(
model_name='gpt-4o',
api_key=os.getenv('OPENAI_API_KEY'),
url='https://api.openai.com/v1'
)
# 获取可用资源模板
resource_templates = await mcp_client.list_resource_templates()
template_dicts = [template.__dict__ for template in resource_templates]
resources_description = json.dumps(template_dicts, ensure_ascii=False)
# 系统提示，指导LLM如何使用资源模板和返回响应
system_message = f'''
你是一个智能助手，能够访问多种数据资源。严格遵循以下协议返回响应：
可用资源模板：{resources_description}
响应规则：
1、当用户询问需要查询数据时，判断是否需要调用资源模板：
- 如果需要查询数据，返回对应的资源URI
- 如果是普通对话，直接回答用户问题
2、资源调用格式：
- 返回纯净的URI字符串，不包含任何其他内容
- URI格式必须严格按照资源模板定义
- 根据用户需求填入合适的参数值
3、URI参数说明：
- date: 日期参数，格式如0101表示1月1日
- limit: 限制返回数据条数
- 其他参数根据资源模板定义填入
4、非数据查询的响应：
- 对于普通对话，直接给出自然语言回答
- 不需要调用资源时，不要返回URI
5、收到资源数据后：
- 将数据转化为用户友好的格式
- 突出显示关键信息
- 保持回复简洁清晰
- 根据用户问题提供相关分析
'''
# 启动聊天会话
chat_session = ChatSession(llm_client=llm_client, mcp_client=mcp_client)
await chat_session.start(system_message=system_message)
if __name__ == "__main__":
asyncio.run(main())
python
客户端最佳实践
- 连接管理：实现自动重连和连接池
- 缓存策略：合理缓存资源和工具响应
- 异步处理：使用异步模式提高性能
- 错误恢复：实现优雅的错误处理和恢复
MCP Sampling原理与优化
Sampling机制深度解析
Sampling原理
MCP 还为我们提供了一个 Sampling 的功能，这个如果从字面来理解会让人摸不到头脑，但实际上这个功能就给了我们一个在执行工具的前后的接口，我们可以在工具执行前后来执行一些操作。比如，当调用本地文件的删除的工具的时候，肯定是期望我们确认后再进行删除。那么，此时就可以使用这个功能。
python
from mcp.types import TextContent
from fastmcp import Context, FastMCP
import psycopg2
from psycopg2.extras import RealDictCursor
import json
mcp = FastMCP("weibo data mcp server",
debug=True,
host="0.0.0.0",
port=8002)
# 数据库连接配置
DB_CONFIG = {
"dbname": "postgres",
"user": "sa",
"password": "123789",
"host": "127.0.0.1",
"port": "15432"
}
def get_db_connection():
"""创建数据库连接"""
return psycopg2.connect(**DB_CONFIG)
@mcp.resource("db://province_info/data/{province_name}")
async def get_province_data(province_name: str) -> str:
"""中国省份（含直辖市）信息
参数:
province_name: 中国省份名称或直辖市，如广东省，北京市
返回:
以 json 格式返回某省份信息
"""
sql = '''
select region_code,
province_name,
capital_city,
area,
population,
abbreviation
from public.chinese_provinces
where province_name = '{province_name}'
'''.format(province_name=province_name)
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql)
rows = cur.fetchall()
return json.dumps(list(rows), default=str)
@mcp.resource("db://movie_info/data/{movie_name}")
async def get_movie_data(movie_name: str) -> str:
"""电影信息
参数:
movie_name: 电影名称
返回:
以 json 格式返回某电影信息
"""
sql = '''
select movie_type,
main_actors,
region,
director,
features,
rating,
movie_name
from public.chinese_movie_ratings
where movie_name = '{movie_name}'
'''.format(movie_name=movie_name)
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql)
rows = cur.fetchall()
return json.dumps(list(rows), default=str)
@mcp.resource("db://weibo_data/data/?date={date}&limit={limit}")
async def get_weibo_data(ctx: Context, date: str, limit: int = 10,) -> str:
"""微博数据
参数:
date: 日期，数据格式为 yyyymmdd
limit: 要求返回的数据量
返回:
以 json 格式返回微博数据
"""
sql = f'''
select weibo_id,
user_name,
publish_date,
publish_time,
weibo_content
from public.t_weibo_ncov
where publish_date = '{date}'
limit {limit}
'''.format(date=date, limit=limit)
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql)
rows = cur.fetchall()
# 通过 Sampling，给每个微博内容增加情感倾向标签
for weibo_data in rows:
system_prompt = '请给出这条微博内容的情感倾向，标注分为三类的其中一个：积极，中性和消极'
response = await ctx.sample(
messages=weibo_data['weibo_content'],
system_prompt=system_prompt
)
assert isinstance(response, TextContent)
weibo_data['sentiment_tag'] = response.text
print(weibo_data)
return json.dumps(list(rows), default=str)
@mcp.resource("db://weibo_data/data/?weibo_user={weibo_user}&limit={limit}")
async def get_weibo_user_data(ctx: Context, weibo_user: str, limit: int = 10,) -> str:
"""某具体微博用户名的微博内容
参数:
weibo_user: 微博用户名
limit: 要求返回的数据量
返回:
以 json 格式返回微博数据
"""
sql = '''
select weibo_id,
user_name,
publish_date,
publish_time,
weibo_content
from public.t_weibo_ncov
where user_name = '{weibo_user}'
limit {limit}
'''.format(weibo_user=weibo_user, limit=limit)
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql)
rows = cur.fetchall()
# 通过 Sampling，给每个微博内容增加情感倾向标签
for weibo_data in rows:
system_prompt = '请给出这条微博内容的情感倾向，标注分为三类的其中一个：积极，中性和消极'
response = await ctx.sample(
messages=weibo_data['weibo_content'],
system_prompt=system_prompt
)
assert isinstance(response, TextContent)
weibo_data['sentiment_tag'] = response.text
print(weibo_data)
return json.dumps(list(rows), default=str)
if __name__ == "__main__":
mcp.run("sse")企业级 MCP 集成方案
企业级MCP服务端
python
#!/usr/bin/env python3
"""
基于SSE协议的MCP计算器服务器
使用FastMCP框架实现，提供计算器工具功能
"""
import asyncio
import json
import logging
import time
import hashlib
import secrets
import os
import uuid
from urllib.parse import urlencode
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Union
from fastapi import FastAPI, Request, HTTPException, Depends, Form, status
from fastapi.responses import StreamingResponse, RedirectResponse, JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
from mcp.server.models import InitializationOptions
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from cachetools import TTLCache, LRUCache
import uvicorn
import math
import operator
import cmath
import statistics
import random
import re
import psutil
import gc
# OAuth 相关 imports
from authlib.jose import jwt
from authlib.common.security import generate_token
from passlib.context import CryptContext
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ================== 并发和缓存优化配置 ==================
# 速率限制器配置
limiter = Limiter(key_func=get_remote_address)
# ================== OAuth 配置和安全设置 ==================
# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# OAuth配置
OAUTH_SECRET_KEY = os.getenv("OAUTH_SECRET_KEY", secrets.token_urlsafe(32))
OAUTH_ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 30
# 创建FastAPI应用
app = FastAPI(
title=" 企业级MCP集成方案",
version="2.1.0",
description="具有OAuth鉴权、速率限制、缓存和并发优化的高性能数学计算器服务"
)
# 添加CORS中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境中应该限制具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 添加速率限制错误处理
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# OAuth2 密码模式
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 授权服务器配置
authorization_server = None # 将在后面初始化
# 创建FastMCP实例
mcp = FastMCP("sse-calculator-server")
# ================== 缓存系统 ==================
# 计算结果缓存 (TTL: 1小时，最大1000个条目)
calculation_cache = TTLCache(maxsize=1000, ttl=3600)
# 历史记录缓存 (LRU: 最大500个条目)
history_cache = LRUCache(maxsize=500)
# 提示词模板缓存 (TTL: 24小时，最大100个条目)
prompt_cache = TTLCache(maxsize=100, ttl=86400)
# ================== 并发控制 ==================
# 异步任务队列
task_queue = asyncio.Queue(maxsize=100)
# 并发控制信号量 (最大20个并发计算任务)
calculation_semaphore = asyncio.Semaphore(20)
# 用户请求频率追踪
user_request_tracker = defaultdict(list)
# ================== 性能监控 ==================
class PerformanceMonitor:
def __init__(self):
self.request_count = 0
self.success_count = 0
self.error_count = 0
self.total_response_time = 0
self.cache_hits = 0
self.cache_misses = 0
self.start_time = time.time()
def record_request(self, response_time: float, success: bool = True, cache_hit: bool = False):
self.request_count += 1
self.total_response_time += response_time
if success:
self.success_count += 1
else:
self.error_count += 1
if cache_hit:
self.cache_hits += 1
else:
self.cache_misses += 1
def get_stats(self) -> Dict[str, Any]:
uptime = time.time() - self.start_time
avg_response_time = self.total_response_time / max(1, self.request_count)
# 获取系统资源使用情况
process = psutil.Process()
memory_info = process.memory_info()
cpu_percent = process.cpu_percent()
return {
"uptime_seconds": round(uptime, 2),
"total_requests": self.request_count,
"success_rate": round(self.success_count / max(1, self.request_count) * 100, 2),
"error_count": self.error_count,
"avg_response_time_ms": round(avg_response_time * 1000, 2),
"cache_hit_rate": round(self.cache_hits / max(1, self.request_count) * 100, 2),
"cache_stats": {
"calculation_cache_size": len(calculation_cache),
"history_cache_size": len(history_cache),
"prompt_cache_size": len(prompt_cache)
},
"system_resources": {
"memory_usage_mb": round(memory_info.rss / 1024 / 1024, 2),
"cpu_percent": cpu_percent,
"active_tasks": len(asyncio.all_tasks())
},
"queue_status": {
"queue_size": task_queue.qsize(),
"available_permits": calculation_semaphore._value
}
}
# 创建性能监控器
performance_monitor = PerformanceMonitor()
# ================== OAuth 数据模型 ==================
class User(BaseModel):
"""用户模型"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
username: str
email: Optional[str] = None
hashed_password: str
is_active: bool = True
is_admin: bool = False
created_at: datetime = Field(default_factory=datetime.now)
scopes: List[str] = Field(default_factory=lambda: ["read"])
class OAuthClient(BaseModel):
"""OAuth客户端模型"""
client_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
client_secret: str = Field(default_factory=lambda: secrets.token_urlsafe(32))
client_name: str
redirect_uris: List[str] = []
grant_types: List[str] = Field(default_factory=lambda: ["authorization_code", "refresh_token"])
response_types: List[str] = Field(default_factory=lambda: ["code"])
scope: str = "read write"
is_active: bool = True
created_at: datetime = Field(default_factory=datetime.now)
created_by: str # user_id
class OAuthToken(BaseModel):
"""OAuth Token模型"""
access_token: str
refresh_token: Optional[str] = None
token_type: str = "Bearer"
expires_in: int
expires_at: datetime
scope: str
client_id: str
user_id: Optional[str] = None
created_at: datetime = Field(default_factory=datetime.now)
class TokenResponse(BaseModel):
"""Token响应模型"""
access_token: str
token_type: str = "Bearer"
expires_in: int
refresh_token: Optional[str] = None
scope: str
class UserCreate(BaseModel):
"""用户创建模型"""
username: str
password: str
email: Optional[str] = None
is_admin: bool = False
class ClientCreate(BaseModel):
"""客户端创建模型"""
client_name: str
redirect_uris: List[str] = []
scope: str = "read write"
class TokenData(BaseModel):
"""Token数据模型"""
user_id: Optional[str] = None
client_id: Optional[str] = None
scopes: List[str] = []
expires_at: Optional[datetime] = None
# ================== OAuth 存储 ==================
# 简单的内存存储（生产环境应使用数据库）
oauth_users: Dict[str, User] = {}
oauth_clients: Dict[str, OAuthClient] = {}
oauth_tokens: Dict[str, OAuthToken] = {}
oauth_authorization_codes: Dict[str, Dict[str, Any]] = {}
revoked_tokens: set = set() # 撤销的令牌列表
# 默认管理员用户
default_admin = User(
username="admin",
hashed_password=pwd_context.hash("admin123"),
is_active=True,
is_admin=True,
email="admin@calculator.com",
scopes=["read", "write", "admin"]
)
oauth_users[default_admin.id] = default_admin
oauth_users["admin"] = default_admin # 允许通过用户名查找
# 默认客户端
default_client = OAuthClient(
client_name="Calculator MCP Client",
redirect_uris=["http://localhost:3000/callback"],
grant_types=["authorization_code", "refresh_token", "client_credentials"],
scope="read write",
created_by=default_admin.id
)
oauth_clients[default_client.client_id] = default_client
# 定义请求模型
class CalculateRequest(BaseModel):
expression: str
class Config:
json_schema_extra = {
"example": {
"expression": "2 + 3 * sin(pi/2)"
}
}
class MathSolveRequest(BaseModel):
problem: str
step_by_step: bool = True
class Config:
json_schema_extra = {
"example": {
"problem": "求解二次方程 x² + 5x + 6 = 0",
"step_by_step": True
}
}
class ExpressionOptimizeRequest(BaseModel):
expression: str
class Config:
json_schema_extra = {
"example": {
"expression": "sqrt(x*x + y*y) + sqrt(x*x + y*y)"
}
}
class ErrorDebugRequest(BaseModel):
expression: str
error: str
class Config:
json_schema_extra = {
"example": {
"expression": "1 / 0",
"error": "ZeroDivisionError: division by zero"
}
}
class MathExplainRequest(BaseModel):
concept: str
level: str = "intermediate"
class Config:
json_schema_extra = {
"example": {
"concept": "导数",
"level": "intermediate"
}
}
class FormulaDeriveRequest(BaseModel):
formula_name: str
context: str = ""
class Config:
json_schema_extra = {
"example": {
"formula_name": "二次公式",
"context": "解一元二次方程"
}
}
# ================== OAuth 认证工具函数 ==================
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""验证密码"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""获取密码哈希"""
return pwd_context.hash(password)
def authenticate_user(username: str, password: str) -> Optional[User]:
"""验证用户身份"""
user = oauth_users.get(username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""创建访问令牌"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
# 转换时间为UNIX时间戳
to_encode.update({
"exp": int(expire.timestamp()),
"iat": int(datetime.utcnow().timestamp())
})
# 使用正确的authlib JWT API
header = {"alg": OAUTH_ALGORITHM}
encoded_jwt = jwt.encode(header, to_encode, OAUTH_SECRET_KEY)
return encoded_jwt.decode('utf-8') if isinstance(encoded_jwt, bytes) else encoded_jwt
def create_refresh_token(data: dict) -> str:
"""创建刷新令牌"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
# 转换时间为UNIX时间戳
to_encode.update({
"exp": int(expire.timestamp()),
"iat": int(datetime.utcnow().timestamp()),
"type": "refresh"
})
# 使用正确的authlib JWT API
header = {"alg": OAUTH_ALGORITHM}
encoded_jwt = jwt.encode(header, to_encode, OAUTH_SECRET_KEY)
return encoded_jwt.decode('utf-8') if isinstance(encoded_jwt, bytes) else encoded_jwt
def verify_token(token: str) -> Optional[TokenData]:
"""验证令牌"""
try:
# 检查令牌是否已被撤销
if token in revoked_tokens:
logger.info(f"令牌已被撤销: {token[:20]}...")
return None
payload = jwt.decode(token, OAUTH_SECRET_KEY)
# 检查是否是刷新令牌
if payload.get("type") == "refresh":
return None
user_id: str = payload.get("sub")
client_id: str = payload.get("client_id")
scopes: List[str] = payload.get("scopes", [])
expires_at = datetime.fromtimestamp(payload.get("exp", 0))
if user_id is None and client_id is None:
return None
token_data = TokenData(
user_id=user_id,
client_id=client_id,
scopes=scopes,
expires_at=expires_at
)
return token_data
except Exception as e:
logger.error(f"Token验证错误: {e}")
return None
def verify_refresh_token(token: str) -> Optional[TokenData]:
"""验证刷新令牌"""
try:
payload = jwt.decode(token, OAUTH_SECRET_KEY)
# 确保是刷新令牌
if payload.get("type") != "refresh":
return None
user_id: str = payload.get("sub")
client_id: str = payload.get("client_id")
scopes: List[str] = payload.get("scopes", [])
expires_at = datetime.fromtimestamp(payload.get("exp", 0))
if user_id is None and client_id is None:
return None
token_data = TokenData(
user_id=user_id,
client_id=client_id,
scopes=scopes,
expires_at=expires_at
)
return token_data
except Exception as e:
logger.error(f"刷新令牌验证错误: {e}")
return None
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
"""获取当前用户"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭据",
headers={"WWW-Authenticate": "Bearer"},
)
token_data = verify_token(token)
if token_data is None:
raise credentials_exception
if token_data.user_id:
user = oauth_users.get(token_data.user_id)
if user is None:
raise credentials_exception
return user
raise credentials_exception
async def get_current_client(token: str = Depends(oauth2_scheme)) -> OAuthClient:
"""获取当前客户端"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证客户端凭据",
headers={"WWW-Authenticate": "Bearer"},
)
token_data = verify_token(token)
if token_data is None:
raise credentials_exception
if token_data.client_id:
client = oauth_clients.get(token_data.client_id)
if client is None:
raise credentials_exception
return client
raise credentials_exception
async def get_current_user_or_client(token: str = Depends(oauth2_scheme)) -> Union[User, OAuthClient]:
"""获取当前用户或客户端"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭据",
headers={"WWW-Authenticate": "Bearer"},
)
token_data = verify_token(token)
if token_data is None:
raise credentials_exception
# 优先返回用户
if token_data.user_id:
user = oauth_users.get(token_data.user_id)
if user is not None:
return user
# 如果没有用户，返回客户端
if token_data.client_id:
client = oauth_clients.get(token_data.client_id)
if client is not None:
return client
raise credentials_exception
def check_scope(required_scopes: List[str]):
"""检查作用域的装饰器"""
def scope_checker(token: str = Depends(oauth2_scheme)):
token_data = verify_token(token)
if token_data is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的访问令牌"
)
# 检查是否有必要的作用域
if not any(scope in token_data.scopes for scope in required_scopes):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"需要以下作用域之一: {', '.join(required_scopes)}"
)
return token_data
return scope_checker
async def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
"""获取管理员用户"""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="需要管理员权限"
)
return current_user
# ================== 缓存助手函数 ==================
def get_cache_key(data: str) -> str:
"""生成缓存键"""
return hashlib.md5(data.encode()).hexdigest()
def clean_expired_requests(user_id: str, window_seconds: int = 60):
"""清理过期的请求记录"""
current_time = time.time()
user_request_tracker[user_id] = [
req_time for req_time in user_request_tracker[user_id]
if current_time - req_time < window_seconds
]
async def check_rate_limit(user_id: str, max_requests: int = 100, window_seconds: int = 60) -> bool:
"""检查用户请求频率"""
clean_expired_requests(user_id, window_seconds)
return len(user_request_tracker[user_id]) < max_requests
def record_user_request(user_id: str):
"""记录用户请求"""
user_request_tracker[user_id].append(time.time())
# ================== 内存优化的历史记录存储 ==================
class OptimizedHistory:
def __init__(self, max_size: int = 1000):
self.max_size = max_size
self._history: List[Dict[str, Any]] = []
self._lock = asyncio.Lock()
async def add_record(self, record: Dict[str, Any]):
"""添加历史记录"""
async with self._lock:
# 直接存储记录副本
self._history.append(record.copy())
# 限制历史记录大小
if len(self._history) > self.max_size:
# 移除最旧的记录
self._history = self._history[-self.max_size:]
# 定期垃圾回收
if len(self._history) % 100 == 0:
gc.collect()
async def get_recent_records(self, limit: int = 50) -> List[Dict[str, Any]]:
"""获取最近的记录"""
async with self._lock:
# 返回最近的记录
return self._history[-limit:] if len(self._history) > limit else self._history.copy()
async def get_stats(self) -> Dict[str, Any]:
"""获取历史记录统计"""
async with self._lock:
total = len(self._history)
return {
"total_records": total,
"valid_records": total,
"memory_efficiency": 100.0
}
# 创建优化的历史记录实例
calculation_history = OptimizedHistory(max_size=1000)
# 数学常数和函数文档
MATH_CONSTANTS_DOC = {
"pi": {"value": math.pi, "description": "圆周率π ≈ 3.14159", "usage": "pi"},
"e": {"value": math.e, "description": "自然常数e ≈ 2.71828", "usage": "e"},
"inf": {"value": float('inf'), "description": "正无穷大", "usage": "float('inf')"},
"nan": {"value": float('nan'), "description": "非数值", "usage": "float('nan')"}
}
MATH_FUNCTIONS_DOC = {
"basic": {
"abs(x)": "返回x的绝对值",
"round(x, n)": "将x四舍五入到n位小数",
"min(...)": "返回参数中的最小值",
"max(...)": "返回参数中的最大值",
"sum(...)": "返回所有参数的和",
"pow(x, y)": "返回x的y次幂"
},
"trigonometric": {
"sin(x)": "正弦函数，x为弧度",
"cos(x)": "余弦函数，x为弧度",
"tan(x)": "正切函数，x为弧度",
"asin(x)": "反正弦函数",
"acos(x)": "反余弦函数",
"atan(x)": "反正切函数"
},
"logarithmic": {
"log(x)": "自然对数，以e为底",
"log10(x)": "常用对数，以10为底",
"log2(x)": "以2为底的对数",
"exp(x)": "e的x次幂"
},
"power_root": {
"sqrt(x)": "平方根",
"pow(x, y)": "x的y次幂",
"pow(x, 1/n)": "x的n次根"
}
}
async def add_to_history(expression: str, result: Any, success: bool, error: str = None):
"""异步添加计算记录到历史"""
history_item = {
"timestamp": datetime.now().isoformat(),
"expression": expression,
"result": result if success else None,
"success": success,
"error": error if not success else None
}
# 使用优化的历史记录存储
await calculation_history.add_record(history_item)
# 缓存最近的历史记录
cache_key = f"history_{len(calculation_history._history)}"
history_cache[cache_key] = history_item
@mcp.tool()
async def calculate(expression: str) -> Dict[str, Any]:
"""
异步计算数学表达式 - 支持缓存和并发优化
支持基本算术运算、三角函数、对数函数等
"""
start_time = time.time()
try:
# 检查缓存
cache_key = get_cache_key(expression)
if cache_key in calculation_cache:
cached_result = calculation_cache[cache_key]
performance_monitor.record_request(
time.time() - start_time,
success=True,
cache_hit=True
)
logger.info(f"Cache hit for expression: {expression}")
return cached_result
# 使用信号量控制并发
async with calculation_semaphore:
# 安全的计算环境 - 扩展的高级数学函数
safe_dict = {
"__builtins__": {},
# 基本函数
"abs": abs,
"round": round,
"min": min,
"max": max,
"sum": sum,
"pow": pow,
"int": int,
"float": float,
"complex": complex,
# 基本数学函数
"sqrt": math.sqrt,
"ceil": math.ceil,
"floor": math.floor,
"trunc": math.trunc,
"fabs": math.fabs,
"factorial": math.factorial,
"gcd": math.gcd,
"lcm": getattr(math, 'lcm', lambda a, b: abs(a * b) // math.gcd(a, b)), # Python 3.9+兼容
# 三角函数
"sin": math.sin,
"cos": math.cos,
"tan": math.tan,
"asin": math.asin,
"acos": math.acos,
"atan": math.atan,
"atan2": math.atan2,
"degrees": math.degrees,
"radians": math.radians,
# 双曲函数
"sinh": math.sinh,
"cosh": math.cosh,
"tanh": math.tanh,
"asinh": math.asinh,
"acosh": math.acosh,
"atanh": math.atanh,
# 对数和指数函数
"log": math.log,
"log10": math.log10,
"log2": math.log2,
"exp": math.exp,
"expm1": math.expm1, # exp(x) - 1
"log1p": math.log1p, # log(1 + x)
# 幂函数和根函数
"pow": pow,
"sqrt": math.sqrt,
"cbrt": lambda x: math.pow(x, 1/3), # 立方根
# 伽马函数和相关函数
"gamma": math.gamma,
"lgamma": math.lgamma,
# 误差函数
"erf": math.erf,
"erfc": math.erfc,
# 统计函数
"mean": statistics.mean,
"median": statistics.median,
"mode": statistics.mode,
"variance": statistics.variance,
"stdev": statistics.stdev,
# 复数函数
"phase": cmath.phase,
"polar": cmath.polar,
"rect": cmath.rect,
# 常数
"pi": math.pi,
"e": math.e,
"tau": math.tau, # 2*pi
"inf": math.inf,
"nan": math.nan,
# 特殊函数
"isfinite": math.isfinite,
"isinf": math.isinf,
"isnan": math.isnan,
"copysign": math.copysign,
"fmod": math.fmod,
"remainder": math.remainder,
"modf": math.modf,
"frexp": math.frexp,
"ldexp": math.ldexp,
}
# 检查表达式安全性
forbidden_chars = ['import', 'exec', 'eval', '__', 'open', 'file']
expression_lower = expression.lower()
for forbidden in forbidden_chars:
if forbidden in expression_lower:
error_msg = f"表达式包含禁用关键字: {forbidden}"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
# 计算表达式
result = eval(expression, safe_dict, {})
# 添加到历史记录
await add_to_history(expression, result, True)
# 缓存结果
result_dict = {
"success": True,
"expression": expression,
"result": result,
"result_type": type(result).__name__,
"cached_at": time.time()
}
calculation_cache[cache_key] = result_dict
# 记录性能指标
performance_monitor.record_request(
time.time() - start_time,
success=True,
cache_hit=False
)
return result_dict
except ZeroDivisionError:
error_msg = "除零错误"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
except OverflowError:
error_msg = "数值溢出"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
except (ValueError, TypeError) as e:
error_msg = f"数值错误: {str(e)}"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
except SyntaxError:
error_msg = "语法错误：请检查表达式格式"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
except Exception as e:
error_msg = f"计算错误: {str(e)}"
await add_to_history(expression, None, False, error_msg)
performance_monitor.record_request(time.time() - start_time, success=False)
return {
"success": False,
"error": error_msg,
"expression": expression
}
@mcp.tool()
def add(a: float, b: float) -> Dict[str, Any]:
"""加法运算"""
result = a + b
return {
"success": True,
"operation": "addition",
"operands": [a, b],
"result": result,
"expression": f"{a} + {b} = {result}"
}
@mcp.tool()
def subtract(a: float, b: float) -> Dict[str, Any]:
"""减法运算"""
result = a - b
return {
"success": True,
"operation": "subtraction",
"operands": [a, b],
"result": result,
"expression": f"{a} - {b} = {result}"
}
@mcp.tool()
def multiply(a: float, b: float) -> Dict[str, Any]:
"""乘法运算"""
result = a * b
return {
"success": True,
"operation": "multiplication",
"operands": [a, b],
"result": result,
"expression": f"{a} × {b} = {result}"
}
@mcp.tool()
def divide(a: float, b: float) -> Dict[str, Any]:
"""除法运算"""
if b == 0:
return {
"success": False,
"operation": "division",
"operands": [a, b],
"error": "除数不能为零",
"expression": f"{a} ÷ {b}"
}
result = a / b
return {
"success": True,
"operation": "division",
"operands": [a, b],
"result": result,
"expression": f"{a} ÷ {b} = {result}"
}
@mcp.tool()
def power(base: float, exponent: float) -> Dict[str, Any]:
"""幂运算"""
try:
result = base ** exponent
return {
"success": True,
"operation": "power",
"operands": [base, exponent],
"result": result,
"expression": f"{base}^{exponent} = {result}"
}
except OverflowError:
return {
"success": False,
"operation": "power",
"operands": [base, exponent],
"error": "结果溢出",
"expression": f"{base}^{exponent}"
}
@mcp.tool()
def square_root(number: float) -> Dict[str, Any]:
"""平方根运算"""
if number < 0:
return {
"success": False,
"operation": "square_root",
"operand": number,
"error": "负数没有实数平方根",
"expression": f"√{number}"
}
result = math.sqrt(number)
return {
"success": True,
"operation": "square_root",
"operand": number,
"result": result,
"expression": f"√{number} = {result}"
}
@mcp.tool()
def trigonometric_functions(function: str, angle: float, unit: str = "radians") -> Dict[str, Any]:
"""
高级三角函数计算
支持正弦、余弦、正切及其反函数，支持弧度和角度单位
"""
try:
# 单位转换
if unit.lower() == "degrees":
if function.startswith("a"): # 反三角函数返回角度
input_angle = math.radians(angle)
else:
input_angle = math.radians(angle)
else:
input_angle = angle
function_map = {
# 三角函数
"sin": math.sin,
"cos": math.cos,
"tan": math.tan,
# 反三角函数
"asin": math.asin,
"acos": math.acos,
"atan": math.atan,
# 双曲函数
"sinh": math.sinh,
"cosh": math.cosh,
"tanh": math.tanh,
# 反双曲函数
"asinh": math.asinh,
"acosh": math.acosh,
"atanh": math.atanh,
}
if function not in function_map:
return {
"success": False,
"error": f"不支持的函数: {function}",
"supported_functions": list(function_map.keys())
}
# 计算结果
if function.startswith("a"): # 反三角函数
if unit.lower() == "degrees":
result = function_map[function](angle)
result_degrees = math.degrees(result)
return {
"success": True,
"function": function,
"input": angle,
"input_unit": unit,
"result": result_degrees,
"result_unit": "degrees",
"result_radians": result,
"expression": f"{function}({angle}) = {result_degrees}°"
}
else:
result = function_map[function](angle)
return {
"success": True,
"function": function,
"input": angle,
"result": result,
"result_unit": "radians",
"expression": f"{function}({angle}) = {result}"
}
else:
result = function_map[function](input_angle)
return {
"success": True,
"function": function,
"input": angle,
"input_unit": unit,
"result": result,
"expression": f"{function}({angle}{'' if unit == 'radians' else '°'}) = {result}"
}
except ValueError as e:
return {
"success": False,
"error": f"数值错误: {str(e)}",
"function": function,
"input": angle
}
except Exception as e:
return {
"success": False,
"error": f"计算错误: {str(e)}",
"function": function,
"input": angle
}
@mcp.tool()
def logarithmic_functions(function: str, value: float, base: Optional[float] = None) -> Dict[str, Any]:
"""
对数函数计算
支持自然对数、常用对数、二进制对数及任意底对数
"""
try:
if function == "ln" or function == "log":
result = math.log(value)
return {
"success": True,
"function": "natural_log",
"input": value,
"result": result,
"expression": f"ln({value}) = {result}"
}
elif function == "log10":
result = math.log10(value)
return {
"success": True,
"function": "common_log",
"input": value,
"result": result,
"expression": f"log₁₀({value}) = {result}"
}
elif function == "log2":
result = math.log2(value)
return {
"success": True,
"function": "binary_log",
"input": value,
"result": result,
"expression": f"log₂({value}) = {result}"
}
elif function == "logn" and base is not None:
result = math.log(value, base)
return {
"success": True,
"function": "arbitrary_base_log",
"input": value,
"base": base,
"result": result,
"expression": f"log_{base}({value}) = {result}"
}
else:
return {
"success": False,
"error": f"不支持的对数函数: {function}",
"supported_functions": ["ln", "log", "log10", "log2", "logn (需要指定底数)"]
}
except ValueError as e:
return {
"success": False,
"error": f"数值错误: {str(e)} (对数的真数必须大于0)",
"function": function,
"input": value
}
except Exception as e:
return {
"success": False,
"error": f"计算错误: {str(e)}",
"function": function,
"input": value
}
@mcp.tool()
def statistical_analysis(data: List[float]) -> Dict[str, Any]:
"""
统计分析工具
计算数据的基本统计量：均值、中位数、众数、方差、标准差等
"""
try:
if not data:
return {
"success": False,
"error": "数据列表不能为空"
}
# 基本统计量
mean_val = statistics.mean(data)
median_val = statistics.median(data)
# 众数（可能有多个或不存在）
try:
mode_val = statistics.mode(data)
except statistics.StatisticsError:
mode_val = None
# 方差和标准差
if len(data) > 1:
variance_val = statistics.variance(data)
stdev_val = statistics.stdev(data)
else:
variance_val = 0
stdev_val = 0
# 其他统计量
min_val = min(data)
max_val = max(data)
range_val = max_val - min_val
count = len(data)
return {
"success": True,
"data": data,
"count": count,
"mean": mean_val,
"median": median_val,
"mode": mode_val,
"variance": variance_val,
"standard_deviation": stdev_val,
"minimum": min_val,
"maximum": max_val,
"range": range_val,
"summary": f"数据分析：{count}个数据点，均值={mean_val:.4f}，中位数={median_val:.4f}，标准差={stdev_val:.4f}"
}
except Exception as e:
return {
"success": False,
"error": f"统计分析错误: {str(e)}",
"data": data
}
@mcp.tool()
def polynomial_evaluation(coefficients: List[float], x: float) -> Dict[str, Any]:
"""
多项式求值
给定多项式系数和自变量值，计算多项式的值
系数从高次项到低次项排列
"""
try:
if not coefficients:
return {
"success": False,
"error": "系数列表不能为空"
}
# 使用霍纳法则求值
result = coefficients[0]
for i in range(1, len(coefficients)):
result = result * x + coefficients[i]
# 构建多项式表达式字符串
n = len(coefficients) - 1
terms = []
for i, coef in enumerate(coefficients):
degree = n - i
if degree == 0:
terms.append(f"{coef}")
elif degree == 1:
if coef == 1:
terms.append("x")
elif coef == -1:
terms.append("-x")
else:
terms.append(f"{coef}x")
else:
if coef == 1:
terms.append(f"x^{degree}")
elif coef == -1:
terms.append(f"-x^{degree}")
else:
terms.append(f"{coef}x^{degree}")
polynomial_str = " + ".join(terms).replace("+ -", "- ")
return {
"success": True,
"coefficients": coefficients,
"x_value": x,
"result": result,
"polynomial": polynomial_str,
"expression": f"P({x}) = {result}",
"evaluation_method": "Horner's method"
}
except Exception as e:
return {
"success": False,
"error": f"多项式求值错误: {str(e)}",
"coefficients": coefficients,
"x_value": x
}
@mcp.tool()
def complex_number_operations(operation: str, z1_real: float, z1_imag: float,
z2_real: Optional[float] = None, z2_imag: Optional[float] = None) -> Dict[str, Any]:
"""
复数运算工具
支持复数的基本运算：加减乘除、模、幅角、共轭等
"""
try:
z1 = complex(z1_real, z1_imag)
if operation in ["add", "subtract", "multiply", "divide"]:
if z2_real is None or z2_imag is None:
return {
"success": False,
"error": "二元运算需要提供第二个复数的实部和虚部"
}
z2 = complex(z2_real, z2_imag)
if operation == "add":
result = z1 + z2
op_symbol = "+"
elif operation == "subtract":
result = z1 - z2
op_symbol = "-"
elif operation == "multiply":
result = z1 * z2
op_symbol = "*"
elif operation == "divide":
if z2 == 0:
return {
"success": False,
"error": "除数不能为零"
}
result = z1 / z2
op_symbol = "/"
elif operation == "modulus" or operation == "abs":
result = abs(z1)
return {
"success": True,
"operation": "modulus",
"input": {"real": z1_real, "imaginary": z1_imag},
"result": result,
"expression": f"|{z1_real}+{z1_imag}i| = {result}"
}
elif operation == "phase" or operation == "argument":
result = cmath.phase(z1)
result_degrees = math.degrees(result)
return {
"success": True,
"operation": "phase",
"input": {"real": z1_real, "imaginary": z1_imag},
"result_radians": result,
"result_degrees": result_degrees,
"expression": f"arg({z1_real}+{z1_imag}i) = {result_degrees}°"
}
elif operation == "conjugate":
result = z1.conjugate()
return {
"success": True,
"operation": "conjugate",
"input": {"real": z1_real, "imaginary": z1_imag},
"result": {"real": result.real, "imaginary": result.imag},
"expression": f"conj({z1_real}+{z1_imag}i) = {result.real}+{result.imag}i"
}
else:
return {
"success": False,
"error": f"不支持的复数运算: {operation}",
"supported_operations": ["add", "subtract", "multiply", "divide", "modulus", "phase", "conjugate"]
}
# 对于二元运算
return {
"success": True,
"operation": operation,
"operand1": {"real": z1_real, "imaginary": z1_imag},
"operand2": {"real": z2_real, "imaginary": z2_imag},
"result": {"real": result.real, "imaginary": result.imag},
"expression": f"({z1_real}+{z1_imag}i) {op_symbol} ({z2_real}+{z2_imag}i) = {result.real}+{result.imag}i"
}
except Exception as e:
return {
"success": False,
"error": f"复数运算错误: {str(e)}",
"operation": operation
}
@mcp.tool()
def expression_parser_advanced(expression: str) -> Dict[str, Any]:
"""
高级数学表达式解析器
解析复杂数学表达式并提供详细信息
"""
try:
# 预处理表达式
expression = expression.replace(" ", "")
# 检查表达式中使用的函数
function_pattern = r'\b([a-zA-Z_][a-zA-Z0-9_]*)\s*\('
functions_used = re.findall(function_pattern, expression)
# 检查表达式中的数字
number_pattern = r'\b\d+\.?\d*\b'
numbers_used = re.findall(number_pattern, expression)
# 检查运算符
operators_used = re.findall(r'[\+\-\*/\^\(\)]', expression)
# 尝试计算表达式
safe_dict = {
"__builtins__": {},
# 添加所有安全的数学函数（复用之前定义的safe_dict内容）
"abs": abs, "round": round, "min": min, "max": max, "sum": sum, "pow": pow,
"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos, "tan": math.tan,
"log": math.log, "log10": math.log10, "exp": math.exp,
"pi": math.pi, "e": math.e, "tau": math.tau,
# 添加更多函数...
}
# 安全性检查
forbidden_chars = ['import', 'exec', 'eval', '__', 'open', 'file']
expression_lower = expression.lower()
for forbidden in forbidden_chars:
if forbidden in expression_lower:
return {
"success": False,
"error": f"表达式包含禁用关键字: {forbidden}",
"expression": expression
}
# 计算结果
result = eval(expression, safe_dict, {})
return {
"success": True,
"original_expression": expression,
"result": result,
"result_type": type(result).__name__,
"functions_used": functions_used,
"numbers_identified": numbers_used,
"operators_used": operators_used,
"complexity_score": len(functions_used) + len(operators_used),
"analysis": {
"function_count": len(functions_used),
"number_count": len(numbers_used),
"operator_count": len(operators_used),
"is_complex": len(functions_used) > 2 or len(operators_used) > 5
}
}
except Exception as e:
return {
"success": False,
"error": f"表达式解析错误: {str(e)}",
"expression": expression
}
@mcp.tool()
def get_calculator_info() -> Dict[str, Any]:
"""获取计算器服务器信息"""
return {
"name": "Advanced SSE Calculator MCP Server",
"version": "2.0.0",
"protocol": "MCP over SSE",
"description": "基于SSE协议的高级MCP计算器服务器，支持复杂数学运算",
"features": [
"基本算术运算 (+, -, ×, ÷)",
"高级三角函数和反三角函数",
"双曲函数和反双曲函数",
"对数和指数函数",
"复数运算",
"统计分析",
"多项式求值",
"高级表达式解析",
"伽马函数和误差函数",
"特殊数学函数"
],
"supported_functions": [
"基本函数: abs, round, min, max, sum, pow, factorial",
"三角函数: sin, cos, tan, asin, acos, atan, degrees, radians",
"双曲函数: sinh, cosh, tanh, asinh, acosh, atanh",
"对数函数: log, log10, log2, exp, expm1, log1p",
"统计函数: mean, median, mode, variance, stdev",
"复数函数: phase, polar, rect",
"特殊函数: gamma, lgamma, erf, erfc",
"数学常数: pi, e, tau, inf, nan"
],
"new_advanced_tools": [
"trigonometric_functions - 高级三角函数计算",
"logarithmic_functions - 对数函数计算",
"statistical_analysis - 统计分析工具",
"polynomial_evaluation - 多项式求值",
"complex_number_operations - 复数运算",
"expression_parser_advanced - 高级表达式解析"
]
}
@mcp.tool()
def list_calculator_tools() -> Dict[str, Any]:
"""列出所有可用的计算器工具"""
tools = [
# 基本运算工具
{
"name": "calculate",
"description": "计算数学表达式（支持高级函数）",
"category": "basic",
"parameters": {
"expression": "string - 数学表达式，如 '2+3*sin(pi/2)', 'gamma(5)', 'log2(8)'"
}
},
{
"name": "add",
"description": "加法运算",
"category": "basic",
"parameters": {
"a": "number - 第一个数",
"b": "number - 第二个数"
}
},
{
"name": "subtract",
"description": "减法运算",
"category": "basic",
"parameters": {
"a": "number - 被减数",
"b": "number - 减数"
}
},
{
"name": "multiply",
"description": "乘法运算",
"category": "basic",
"parameters": {
"a": "number - 第一个因数",
"b": "number - 第二个因数"
}
},
{
"name": "divide",
"description": "除法运算",
"category": "basic",
"parameters": {
"a": "number - 被除数",
"b": "number - 除数"
}
},
{
"name": "power",
"description": "幂运算",
"category": "basic",
"parameters": {
"base": "number - 底数",
"exponent": "number - 指数"
}
},
{
"name": "square_root",
"description": "平方根运算",
"category": "basic",
"parameters": {
"number": "number - 被开方数"
}
},
# 高级数学函数工具
{
"name": "trigonometric_functions",
"description": "高级三角函数计算",
"category": "advanced_trigonometry",
"parameters": {
"function": "string - 函数名 (sin, cos, tan, asin, acos, atan, sinh, cosh, tanh, asinh, acosh, atanh)",
"angle": "number - 角度值",
"unit": "string - 角度单位 ('radians' 或 'degrees'), 默认 'radians'"
}
},
{
"name": "logarithmic_functions",
"description": "对数函数计算",
"category": "advanced_logarithm",
"parameters": {
"function": "string - 对数类型 ('ln', 'log', 'log10', 'log2', 'logn')",
"value": "number - 真数",
"base": "number - 底数 (仅对 'logn' 需要), 可选"
}
},
{
"name": "statistical_analysis",
"description": "统计分析工具",
"category": "statistics",
"parameters": {
"data": "array of numbers - 数据列表，如 [1, 2, 3, 4, 5]"
}
},
{
"name": "polynomial_evaluation",
"description": "多项式求值",
"category": "algebra",
"parameters": {
"coefficients": "array of numbers - 多项式系数（高次到低次），如 [1, -2, 1] 表示 x²-2x+1",
"x": "number - 自变量值"
}
},
{
"name": "complex_number_operations",
"description": "复数运算工具",
"category": "complex",
"parameters": {
"operation": "string - 运算类型 ('add', 'subtract', 'multiply', 'divide', 'modulus', 'phase', 'conjugate')",
"z1_real": "number - 第一个复数的实部",
"z1_imag": "number - 第一个复数的虚部",
"z2_real": "number - 第二个复数的实部 (二元运算时需要), 可选",
"z2_imag": "number - 第二个复数的虚部 (二元运算时需要), 可选"
}
},
{
"name": "expression_parser_advanced",
"description": "高级数学表达式解析器",
"category": "parsing",
"parameters": {
"expression": "string - 要解析的数学表达式"
}
}
]
# 按类别分组
categories = {}
for tool in tools:
category = tool["category"]
if category not in categories:
categories[category] = []
categories[category].append(tool["name"])
return {
"total_tools": len(tools),
"tools": tools,
"categories": categories,
"advanced_features": [
"支持60+个数学函数",
"三角函数和双曲函数完整支持",
"任意底对数计算",
"复数运算",
"统计分析",
"多项式求值",
"表达式解析和分析",
"角度和弧度单位转换"
]
}
# ================== Resources 资源 ==================
@mcp.resource("file://calculator/history")
def get_calculation_history() -> str:
"""获取计算历史记录"""
if not calculation_history:
return "暂无计算历史记录"
history_text = "# 计算历史记录\n\n"
for i, record in enumerate(calculation_history[-50:], 1): # 只显示最近50条
timestamp = record["timestamp"].split('T')[1][:8] # 只显示时间部分
if record["success"]:
history_text += f"{i}. [{timestamp}] {record['expression']} = {record['result']}\n"
else:
history_text += f"{i}. [{timestamp}] {record['expression']} ❌ {record['error']}\n"
return history_text
@mcp.resource("file://calculator/functions")
def get_math_functions_doc() -> str:
"""获取数学函数文档"""
doc_text = "# 数学函数文档\n\n"
for category, functions in MATH_FUNCTIONS_DOC.items():
doc_text += f"## {category.title()} Functions\n\n"
for func, desc in functions.items():
doc_text += f"- **{func}**: {desc}\n"
doc_text += "\n"
return doc_text
@mcp.resource("file://calculator/constants")
def get_math_constants_doc() -> str:
"""获取数学常数文档"""
doc_text = "# 数学常数文档\n\n"
for name, info in MATH_CONSTANTS_DOC.items():
doc_text += f"- **{name}**: {info['description']}\n"
doc_text += f" - 值: {info['value']}\n"
doc_text += f" - 用法: {info['usage']}\n\n"
return doc_text
@mcp.resource("file://calculator/examples")
def get_calculation_examples() -> str:
"""获取计算示例"""
examples_text = """# 计算示例
## 基本算术运算
- 加法: `2 + 3 = 5`
- 减法: `10 - 4 = 6`
- 乘法: `5 * 6 = 30`
- 除法: `15 / 3 = 5`
- 幂运算: `2 ** 8 = 256`
## 三角函数
- 正弦: `sin(pi/2) = 1.0`
- 余弦: `cos(0) = 1.0`
- 正切: `tan(pi/4) = 1.0`
## 对数函数
- 自然对数: `log(e) = 1.0`
- 常用对数: `log10(100) = 2.0`
- 指数: `exp(1) = 2.718...`
## 复杂表达式
- 圆的面积: `pi * pow(radius, 2)`
- 二次方程: `(-b + sqrt(b**2 - 4*a*c)) / (2*a)`
- 统计: `(sum([1,2,3,4,5])) / 5 = 3.0`
## 数学常数
- π (pi): `3.141592653589793`
- e: `2.718281828459045`
- 黄金比例: `(1 + sqrt(5)) / 2 = 1.618...`
"""
return examples_text
@mcp.resource("file://calculator/status")
def get_calculator_status() -> str:
"""获取计算器状态"""
total_calculations = len(calculation_history)
successful = sum(1 for record in calculation_history if record["success"])
failed = total_calculations - successful
if total_calculations > 0:
success_rate = (successful / total_calculations) * 100
recent_activity = calculation_history[-5:] if calculation_history else []
else:
success_rate = 0
recent_activity = []
status_text = f"""# 计算器状态
## 统计信息
- 总计算次数: {total_calculations}
- 成功次数: {successful}
- 失败次数: {failed}
- 成功率: {success_rate:.1f}%
## 最近活动
"""
if recent_activity:
for record in recent_activity:
timestamp = record["timestamp"].split('T')[1][:8]
status = "✅" if record["success"] else "❌"
status_text += f"- [{timestamp}] {record['expression']} {status}\n"
else:
status_text += "- 暂无活动记录\n"
return status_text
# ================== Prompts 提示词模板 ==================
@mcp.prompt("math-solve")
def math_problem_solver(problem: str, step_by_step: bool = True) -> str:
"""数学问题解决提示词模板"""
template = f"""请解决以下数学问题：
问题: {problem}
{"请提供详细的解题步骤：" if step_by_step else ""}
解题要求：
1. 理解问题并确定解题方法
2. {"逐步展示计算过程" if step_by_step else "直接给出答案"}
3. 检验答案的合理性
4. 如果可能，提供替代解法
可用的数学函数包括：
- 基本运算：+, -, *, /, **, sqrt()
- 三角函数：sin(), cos(), tan()
- 对数函数：log(), log10(), exp()
- 其他：abs(), round(), min(), max()
- 常数：pi, e
请开始解题：
"""
return template
@mcp.prompt("expression-optimize")
def expression_optimizer(expression: str) -> str:
"""数学表达式优化提示词模板"""
template = f"""请优化以下数学表达式：
原表达式: {expression}
优化目标：
1. 简化表达式结构
2. 减少计算复杂度
3. 提高数值稳定性
4. 保持数学等价性
优化策略：
- 合并同类项
- 提取公因子
- 使用恒等变换
- 避免数值溢出
- 减少函数调用次数
请提供：
1. 优化后的表达式
2. 优化理由和步骤
3. 数值验证对比
开始优化：
"""
return template
@mcp.prompt("error-debug")
def calculation_error_debug(expression: str, error: str) -> str:
"""计算错误调试提示词模板"""
template = f"""计算表达式时遇到错误，请帮助调试：
错误表达式: {expression}
错误信息: {error}
调试检查项：
1. 语法错误检查
- 括号匹配
- 运算符使用
- 函数名拼写
2. 数学错误检查
- 除零错误
- 定义域问题
- 数值溢出
3. 类型错误检查
- 参数类型
- 返回值类型
- 类型转换
4. 逻辑错误检查
- 运算顺序
- 优先级问题
- 边界情况
请提供：
1. 错误原因分析
2. 修正建议
3. 正确的表达式
4. 预防措施
开始调试：
"""
return template
@mcp.prompt("math-explain")
def math_concept_explain(concept: str, level: str = "intermediate") -> str:
"""数学概念解释提示词模板"""
level_desc = {
"basic": "基础水平（初中数学）",
"intermediate": "中等水平（高中数学）",
"advanced": "高等水平（大学数学）"
}
template = f"""请解释以下数学概念：
概念: {concept}
解释水平: {level_desc.get(level, "中等水平")}
解释内容包括：
1. 概念定义
- 严格的数学定义
- 通俗易懂的解释
2. 几何或代数意义
- 直观理解
- 图形表示（如果适用）
3. 计算方法
- 公式推导
- 计算技巧
4. 实际应用
- 生活中的例子
- 工程应用
5. 相关概念
- 关联知识点
- 扩展学习
请开始解释：
"""
return template
@mcp.prompt("formula-derive")
def formula_derivation(formula_name: str, context: str = "") -> str:
"""公式推导提示词模板"""
template = f"""请推导以下数学公式：
公式名称: {formula_name}
{"相关背景: " + context if context else ""}
推导要求：
1. 从基本原理出发
2. 逐步推导过程
3. 说明每一步的依据
4. 标注关键变换
推导框架：
1. 问题设定
- 定义变量
- 设定条件
- 明确目标
2. 理论基础
- 相关定理
- 基本公式
- 数学工具
3. 推导过程
- 逐步变换
- 说明理由
- 检查合理性
4. 结果验证
- 特殊情况检验
- 量纲分析
- 物理意义
开始推导：
"""
return template
# ================== OAuth 授权服务器端点 ==================
@app.post("/token", response_model=TokenResponse)
async def get_access_token(
username: str = Form(...),
password: str = Form(...),
grant_type: str = Form(...),
scope: Optional[str] = Form(None)
):
"""
OAuth2 Token端点 - 支持密码模式和客户端凭据模式
"""
try:
# 检查grant_type
if grant_type == "password":
# 尝试用户认证（密码模式）
user = authenticate_user(username, password)
if user:
# 用户认证成功
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
scopes = scope.split() if scope else user.scopes
access_token_data = {
"sub": user.id,
"scopes": scopes,
"token_type": "access"
}
refresh_token_data = {
"sub": user.id,
"scopes": scopes,
"token_type": "refresh"
}
access_token = create_access_token(data=access_token_data, expires_delta=access_token_expires)
refresh_token = create_refresh_token(data=refresh_token_data)
# 存储token信息
token_info = OAuthToken(
access_token=access_token,
refresh_token=refresh_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
expires_at=datetime.utcnow() + access_token_expires,
scope=" ".join(scopes),
client_id="", # 密码模式没有客户端
user_id=user.id
)
oauth_tokens[access_token] = token_info
return TokenResponse(
access_token=access_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
refresh_token=refresh_token,
scope=" ".join(scopes)
)
elif grant_type == "client_credentials":
# 尝试客户端认证
for client_id, client in oauth_clients.items():
if client.client_id == username and client.client_secret == password:
# 客户端认证成功
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
scopes = scope.split() if scope else client.scope.split()
access_token_data = {
"client_id": client.client_id,
"scopes": scopes,
"token_type": "access"
}
access_token = create_access_token(data=access_token_data, expires_delta=access_token_expires)
# 存储token信息
token_info = OAuthToken(
access_token=access_token,
refresh_token=None, # 客户端凭据模式通常不提供刷新令牌
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
expires_at=datetime.utcnow() + access_token_expires,
scope=" ".join(scopes),
client_id=client.client_id,
user_id=None
)
oauth_tokens[access_token] = token_info
return TokenResponse(
access_token=access_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
scope=" ".join(scopes)
)
# 认证失败
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Token获取错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="服务器内部错误"
)
@app.post("/refresh", response_model=TokenResponse)
async def refresh_access_token(refresh_token: str = Form(...)):
"""
刷新访问令牌
"""
try:
# 验证刷新令牌
token_data = verify_refresh_token(refresh_token)
if token_data is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的刷新令牌"
)
# 检查用户是否还存在且活跃
if token_data.user_id:
user = oauth_users.get(token_data.user_id)
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户不存在或已被禁用"
)
# 创建新的访问令牌
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token_data = {
"sub": user.id,
"scopes": token_data.scopes,
"token_type": "access"
}
access_token = create_access_token(data=access_token_data, expires_delta=access_token_expires)
# 可选：创建新的刷新令牌
refresh_token_data = {
"sub": user.id,
"scopes": token_data.scopes,
"token_type": "refresh"
}
new_refresh_token = create_refresh_token(data=refresh_token_data)
# 存储新token信息
token_info = OAuthToken(
access_token=access_token,
refresh_token=new_refresh_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
expires_at=datetime.utcnow() + access_token_expires,
scope=" ".join(token_data.scopes),
client_id="",
user_id=user.id
)
oauth_tokens[access_token] = token_info
return TokenResponse(
access_token=access_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
refresh_token=new_refresh_token,
scope=" ".join(token_data.scopes)
)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的刷新令牌"
)
except HTTPException:
raise
except Exception as e:
logger.error(f"令牌刷新错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="服务器内部错误"
)
@app.get("/authorize")
async def authorize(
response_type: str,
client_id: str,
redirect_uri: str,
scope: Optional[str] = None,
state: Optional[str] = None,
code_challenge: Optional[str] = None,
code_challenge_method: Optional[str] = None
):
"""
OAuth2 授权端点 - Authorization Code流程
"""
try:
# 验证客户端
client = oauth_clients.get(client_id)
if not client:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的客户端ID"
)
# 验证重定向URI
if redirect_uri not in client.redirect_uris:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的重定向URI"
)
# 验证响应类型
if response_type not in client.response_types:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="不支持的响应类型"
)
# 生成授权码
authorization_code = generate_token(32)
code_data = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"scope": scope or client.scope,
"code_challenge": code_challenge,
"code_challenge_method": code_challenge_method,
"expires_at": datetime.utcnow() + timedelta(minutes=10), # 授权码10分钟过期
"used": False
}
oauth_authorization_codes[authorization_code] = code_data
# 构建重定向URL
params = {"code": authorization_code}
if state:
params["state"] = state
redirect_url = f"{redirect_uri}?{urlencode(params)}"
return RedirectResponse(url=redirect_url)
except HTTPException:
raise
except Exception as e:
logger.error(f"授权端点错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="服务器内部错误"
)
@app.post("/token/authorization_code", response_model=TokenResponse)
async def exchange_authorization_code(
grant_type: str = Form(...),
code: str = Form(...),
redirect_uri: str = Form(...),
client_id: str = Form(...),
client_secret: str = Form(...),
code_verifier: Optional[str] = Form(None)
):
"""
使用授权码交换访问令牌
"""
try:
# 验证授权类型
if grant_type != "authorization_code":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="不支持的授权类型"
)
# 验证授权码
code_data = oauth_authorization_codes.get(code)
if not code_data or code_data["used"]:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的授权码"
)
# 检查授权码是否过期
if datetime.utcnow() > code_data["expires_at"]:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="授权码已过期"
)
# 验证客户端
client = oauth_clients.get(client_id)
if not client or client.client_secret != client_secret:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="客户端认证失败"
)
# 验证重定向URI
if code_data["redirect_uri"] != redirect_uri:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="重定向URI不匹配"
)
# PKCE验证
if code_data.get("code_challenge"):
if not code_verifier:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="缺少code_verifier"
)
# 这里应该实现PKCE验证逻辑
# 简化实现，实际应该根据code_challenge_method验证
# 标记授权码为已使用
oauth_authorization_codes[code]["used"] = True
# 创建访问令牌（这里需要实际的用户信息，简化为使用默认管理员）
user = default_admin # 实际应该从授权流程中获取用户信息
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
scopes = code_data["scope"].split() if isinstance(code_data["scope"], str) else [code_data["scope"]]
access_token_data = {
"sub": user.id,
"client_id": client_id,
"scopes": scopes,
"token_type": "access"
}
refresh_token_data = {
"sub": user.id,
"client_id": client_id,
"scopes": scopes,
"token_type": "refresh"
}
access_token = create_access_token(data=access_token_data, expires_delta=access_token_expires)
refresh_token = create_refresh_token(data=refresh_token_data)
# 存储token信息
token_info = OAuthToken(
access_token=access_token,
refresh_token=refresh_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
expires_at=datetime.utcnow() + access_token_expires,
scope=" ".join(scopes),
client_id=client_id,
user_id=user.id
)
oauth_tokens[access_token] = token_info
return TokenResponse(
access_token=access_token,
token_type="Bearer",
expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
refresh_token=refresh_token,
scope=" ".join(scopes)
)
except HTTPException:
raise
except Exception as e:
logger.error(f"授权码交换错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="服务器内部错误"
)
@app.get("/oauth/info")
async def get_oauth_info():
"""获取OAuth服务器信息"""
return {
"issuer": "https://localhost:8000",
"authorization_endpoint": "/authorize",
"token_endpoint": "/token",
"refresh_endpoint": "/refresh",
"supported_grant_types": [
"authorization_code",
"client_credentials",
"password",
"refresh_token"
],
"supported_response_types": ["code", "token"],
"supported_scopes": ["read", "write", "admin"],
"token_endpoint_auth_methods_supported": [
"client_secret_basic",
"client_secret_post"
],
"code_challenge_methods_supported": ["S256", "plain"]
}
# ================== OAuth 客户端和用户管理端点 ==================
@app.post("/oauth/clients", status_code=201)
async def create_oauth_client(
client_data: ClientCreate,
admin_user: User = Depends(get_admin_user)
):
"""创建OAuth客户端（仅管理员）"""
try:
# 创建新客户端
new_client = OAuthClient(
client_name=client_data.client_name,
redirect_uris=client_data.redirect_uris,
scope=client_data.scope,
created_by=admin_user.id
)
# 存储客户端
oauth_clients[new_client.client_id] = new_client
return {
"client_id": new_client.client_id,
"client_secret": new_client.client_secret,
"client_name": new_client.client_name,
"redirect_uris": new_client.redirect_uris,
"scope": new_client.scope,
"created_at": new_client.created_at.isoformat(),
"message": "客户端创建成功"
}
except Exception as e:
logger.error(f"客户端创建错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="客户端创建失败"
)
@app.get("/oauth/clients")
async def list_oauth_clients(
admin_user: User = Depends(get_admin_user)
):
"""列出所有OAuth客户端（仅管理员）"""
try:
clients = []
for client in oauth_clients.values():
clients.append({
"client_id": client.client_id,
"client_name": client.client_name,
"redirect_uris": client.redirect_uris,
"scope": client.scope,
"is_active": client.is_active,
"created_at": client.created_at.isoformat(),
"created_by": client.created_by
})
return {
"clients": clients,
"total": len(clients)
}
except Exception as e:
logger.error(f"客户端列表获取错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="客户端列表获取失败"
)
@app.get("/oauth/clients/{client_id}")
async def get_oauth_client(
client_id: str,
admin_user: User = Depends(get_admin_user)
):
"""获取OAuth客户端详情（仅管理员）"""
try:
client = oauth_clients.get(client_id)
if not client:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="客户端不存在"
)
return {
"client_id": client.client_id,
"client_name": client.client_name,
"redirect_uris": client.redirect_uris,
"grant_types": client.grant_types,
"response_types": client.response_types,
"scope": client.scope,
"is_active": client.is_active,
"created_at": client.created_at.isoformat(),
"created_by": client.created_by
}
except HTTPException:
raise
except Exception as e:
logger.error(f"客户端获取错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="客户端获取失败"
)
@app.put("/oauth/clients/{client_id}")
async def update_oauth_client(
client_id: str,
client_data: ClientCreate,
admin_user: User = Depends(get_admin_user)
):
"""更新OAuth客户端（仅管理员）"""
try:
client = oauth_clients.get(client_id)
if not client:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="客户端不存在"
)
# 更新客户端信息
client.client_name = client_data.client_name
client.redirect_uris = client_data.redirect_uris
client.scope = client_data.scope
return {
"client_id": client.client_id,
"client_name": client.client_name,
"redirect_uris": client.redirect_uris,
"scope": client.scope,
"message": "客户端更新成功"
}
except HTTPException:
raise
except Exception as e:
logger.error(f"客户端更新错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="客户端更新失败"
)
@app.delete("/oauth/clients/{client_id}")
async def delete_oauth_client(
client_id: str,
admin_user: User = Depends(get_admin_user)
):
"""删除OAuth客户端（仅管理员）"""
try:
if client_id not in oauth_clients:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="客户端不存在"
)
# 删除客户端
deleted_client = oauth_clients.pop(client_id)
# 删除相关的令牌（可选）
tokens_to_delete = []
for token_key, token in oauth_tokens.items():
if token.client_id == client_id:
tokens_to_delete.append(token_key)
for token_key in tokens_to_delete:
oauth_tokens.pop(token_key, None)
return {
"message": f"客户端 '{deleted_client.client_name}' 删除成功",
"deleted_tokens": len(tokens_to_delete)
}
except HTTPException:
raise
except Exception as e:
logger.error(f"客户端删除错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="客户端删除失败"
)
@app.post("/oauth/users", status_code=201)
async def create_user(
user_data: UserCreate,
admin_user: User = Depends(get_admin_user)
):
"""创建用户（仅管理员）"""
try:
# 检查用户名是否已存在
for existing_user in oauth_users.values():
if existing_user.username == user_data.username:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="用户名已存在"
)
# 创建新用户
hashed_password = get_password_hash(user_data.password)
new_user = User(
username=user_data.username,
email=user_data.email,
hashed_password=hashed_password,
is_admin=user_data.is_admin,
scopes=["read"] + (["write", "admin"] if user_data.is_admin else ["write"])
)
# 存储用户
oauth_users[new_user.id] = new_user
oauth_users[new_user.username] = new_user # 允许通过用户名查找
return {
"user_id": new_user.id,
"username": new_user.username,
"email": new_user.email,
"is_admin": new_user.is_admin,
"is_active": new_user.is_active,
"scopes": new_user.scopes,
"created_at": new_user.created_at.isoformat(),
"message": "用户创建成功"
}
except HTTPException:
raise
except Exception as e:
logger.error(f"用户创建错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="用户创建失败"
)
@app.get("/oauth/users")
async def list_users(
admin_user: User = Depends(get_admin_user)
):
"""列出所有用户（仅管理员）"""
try:
users = []
seen_users = set()
for user in oauth_users.values():
# 避免重复（因为用户同时以ID和用户名为键存储）
if user.id not in seen_users:
users.append({
"user_id": user.id,
"username": user.username,
"email": user.email,
"is_admin": user.is_admin,
"is_active": user.is_active,
"scopes": user.scopes,
"created_at": user.created_at.isoformat()
})
seen_users.add(user.id)
return {
"users": users,
"total": len(users)
}
except Exception as e:
logger.error(f"用户列表获取错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="用户列表获取失败"
)
@app.get("/oauth/me")
async def get_current_user_info(
current_user: User = Depends(get_current_user)
):
"""获取当前用户信息"""
return {
"user_id": current_user.id,
"username": current_user.username,
"email": current_user.email,
"is_admin": current_user.is_admin,
"is_active": current_user.is_active,
"scopes": current_user.scopes,
"created_at": current_user.created_at.isoformat()
}
@app.get("/oauth/tokens")
async def list_active_tokens(
admin_user: User = Depends(get_admin_user)
):
"""列出所有活跃的令牌（仅管理员）"""
try:
active_tokens = []
current_time = datetime.utcnow()
for token_key, token in oauth_tokens.items():
if token.expires_at > current_time:
active_tokens.append({
"access_token": token_key[:20] + "...", # 只显示前20字符
"token_type": token.token_type,
"expires_at": token.expires_at.isoformat(),
"scope": token.scope,
"client_id": token.client_id or "N/A",
"user_id": token.user_id or "N/A",
"created_at": token.created_at.isoformat()
})
return {
"active_tokens": active_tokens,
"total": len(active_tokens)
}
except Exception as e:
logger.error(f"令牌列表获取错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="令牌列表获取失败"
)
@app.post("/oauth/revoke")
async def revoke_token(
token: str = Form(...),
current_user: User = Depends(get_current_user)
):
"""撤销令牌"""
try:
# 查找并撤销令牌
token_found = False
for token_key in list(oauth_tokens.keys()):
token_info = oauth_tokens[token_key]
if token_key == token or token_info.refresh_token == token:
# 检查权限：用户只能撤销自己的令牌，管理员可以撤销任何令牌
if token_info.user_id == current_user.id or current_user.is_admin:
# 将令牌添加到撤销列表
revoked_tokens.add(token_key)
if token_info.refresh_token:
revoked_tokens.add(token_info.refresh_token)
# 从存储中删除令牌
oauth_tokens.pop(token_key, None)
token_found = True
logger.info(f"令牌已撤销: {token_key[:20]}...")
break
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="无权撤销此令牌"
)
if not token_found:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="令牌不存在或已过期"
)
return {"message": "令牌撤销成功"}
except HTTPException:
raise
except Exception as e:
logger.error(f"令牌撤销错误: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="令牌撤销失败"
)
# SSE端点
@app.get("/sse")
async def sse_endpoint(request: Request):
"""SSE连接端点"""
async def event_stream():
try:
# 发送连接确认
yield f"data: {json.dumps({'type': 'connection', 'status': 'connected', 'server': 'sse-calculator-server'})}\n\n"
# 发送服务器信息
server_info = get_calculator_info()
yield f"data: {json.dumps({'type': 'server_info', 'data': server_info})}\n\n"
# 保持连接活跃
while True:
if await request.is_disconnected():
break
# 发送心跳
yield f"data: {json.dumps({'type': 'heartbeat', 'timestamp': asyncio.get_event_loop().time()})}\n\n"
await asyncio.sleep(30) # 每30秒发送一次心跳
except asyncio.CancelledError:
logger.info("SSE connection cancelled")
except Exception as e:
logger.error(f"SSE stream error: {e}")
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Access-Control-Allow-Origin": "*",
}
)
@app.post("/calculate")
@limiter.limit("60/minute") # 每分钟最多60次请求
async def calculate_endpoint(
request: Request,
request_data: CalculateRequest,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""
优化的计算数学表达式接口 - 支持速率限制和并发控制
支持的数学函数：
- 基本运算：+, -, *, /, **, sqrt()
- 三角函数：sin(), cos(), tan()
- 对数函数：log(), log10(), exp()
- 其他函数：abs(), round(), min(), max()
- 数学常数：pi, e
示例：
- 简单计算：2 + 3 * 4
- 三角函数：sin(pi/2)
- 复杂表达式：sqrt(pow(3, 2) + pow(4, 2))
优化特性：
- 智能缓存：相同表达式直接返回缓存结果
- 并发控制：限制同时计算任务数量
- 性能监控：记录响应时间和成功率
"""
try:
expression = request_data.expression.strip()
if not expression:
raise HTTPException(status_code=400, detail="表达式不能为空")
# 获取客户端IP进行额外的用户级别限制
client_ip = get_remote_address(request)
# 检查用户级别的请求频率（每分钟100次）
if not await check_rate_limit(client_ip, max_requests=100, window_seconds=60):
raise HTTPException(
status_code=429,
detail="请求过于频繁，请稍后再试"
)
# 记录用户请求
record_user_request(client_ip)
# 调用优化的异步计算函数
result = await calculate(expression)
return result
except HTTPException:
# 重新抛出HTTP异常
raise
except Exception as e:
logger.error(f"计算接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
# ========== 高级数学函数HTTP端点 ==========
@app.post("/trigonometric_functions")
async def trigonometric_functions_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""高级三角函数计算接口"""
try:
data = await request.json()
function = data.get("function", "")
angle = data.get("angle", 0)
unit = data.get("unit", "radians")
if not function:
raise HTTPException(status_code=400, detail="函数名不能为空")
result = trigonometric_functions(function, float(angle), unit)
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"三角函数计算接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/logarithmic_functions")
async def logarithmic_functions_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""对数函数计算接口"""
try:
data = await request.json()
function = data.get("function", "")
value = data.get("value", 0)
base = data.get("base")
if not function:
raise HTTPException(status_code=400, detail="函数名不能为空")
result = logarithmic_functions(function, float(value), float(base) if base is not None else None)
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"对数函数计算接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/statistical_analysis")
async def statistical_analysis_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""统计分析接口"""
try:
data = await request.json()
data_list = data.get("data", [])
if not data_list:
raise HTTPException(status_code=400, detail="数据列表不能为空")
result = statistical_analysis([float(x) for x in data_list])
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"统计分析接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/polynomial_evaluation")
async def polynomial_evaluation_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""多项式求值接口"""
try:
data = await request.json()
coefficients = data.get("coefficients", [])
x = data.get("x", 0)
if not coefficients:
raise HTTPException(status_code=400, detail="系数列表不能为空")
result = polynomial_evaluation([float(c) for c in coefficients], float(x))
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"多项式求值接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/complex_number_operations")
async def complex_number_operations_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""复数运算接口"""
try:
data = await request.json()
operation = data.get("operation", "")
z1_real = data.get("z1_real", 0)
z1_imag = data.get("z1_imag", 0)
z2_real = data.get("z2_real")
z2_imag = data.get("z2_imag")
if not operation:
raise HTTPException(status_code=400, detail="运算类型不能为空")
result = complex_number_operations(
operation,
float(z1_real),
float(z1_imag),
float(z2_real) if z2_real is not None else None,
float(z2_imag) if z2_imag is not None else None
)
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"复数运算接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/expression_parser_advanced")
async def expression_parser_advanced_endpoint(
request: Request,
current_user_or_client: Union[User, OAuthClient] = Depends(get_current_user_or_client),
_token_data: TokenData = Depends(check_scope(["read", "write"]))
):
"""高级数学表达式解析接口"""
try:
data = await request.json()
expression = data.get("expression", "")
if not expression:
raise HTTPException(status_code=400, detail="表达式不能为空")
result = expression_parser_advanced(expression)
return result
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="无效的JSON格式")
except Exception as e:
logger.error(f"表达式解析接口错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""简化的健康检查端点"""
try:
return {
"status": "healthy",
"server": "sse-calculator-server-oauth",
"version": "2.1.0",
"timestamp": time.time(),
"cache_size": len(calculation_cache),
"history_size": len(calculation_history._history) if hasattr(calculation_history, '_history') else 0
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"server": "sse-calculator-server-oauth",
"version": "2.1.0"
}
@app.get("/performance")
@limiter.limit("30/minute")
async def get_performance_stats(
request: Request = None,
current_user: User = Depends(get_current_user),
_token_data: TokenData = Depends(check_scope(["read", "admin"]))
):
"""获取详细的性能统计信息"""
try:
# 获取基础性能统计
stats = performance_monitor.get_stats()
# 获取历史记录统计
history_stats = await calculation_history.get_stats()
# 合并统计信息
stats["history_stats"] = history_stats
stats["cache_details"] = {
"calculation_cache": {
"size": len(calculation_cache),
"maxsize": calculation_cache.maxsize,
"ttl": getattr(calculation_cache, 'ttl', 'N/A')
},
"history_cache": {
"size": len(history_cache),
"maxsize": history_cache.maxsize
},
"prompt_cache": {
"size": len(prompt_cache),
"maxsize": prompt_cache.maxsize,
"ttl": getattr(prompt_cache, 'ttl', 'N/A')
}
}
return {
"status": "success",
"data": stats,
"timestamp": time.time()
}
except Exception as e:
logger.error(f"性能统计获取错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/resources/history")
@limiter.limit("120/minute")
async def get_history_api(request: Request):
"""获取计算历史记录API - 优化版本"""
try:
# 使用优化的历史记录获取
recent_records = await calculation_history.get_recent_records(limit=50)
if not recent_records:
return {"content": "暂无计算历史记录"}
history_text = "# 计算历史记录 (最近50条)\n\n"
for i, record in enumerate(recent_records, 1):
timestamp = record["timestamp"].split('T')[1][:8] # 只显示时间部分
if record["success"]:
history_text += f"{i}. [{timestamp}] {record['expression']} = {record['result']}\n"
else:
history_text += f"{i}. [{timestamp}] {record['expression']} ❌ {record['error']}\n"
return {
"content": history_text,
"total_records": len(recent_records),
"cached": True
}
except Exception as e:
logger.error(f"历史记录API错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/resources/functions")
async def get_functions_api():
"""获取数学函数文档API"""
return {"content": get_math_functions_doc()}
@app.get("/api/resources/constants")
async def get_constants_api():
"""获取数学常数文档API"""
return {"content": get_math_constants_doc()}
@app.get("/api/resources/examples")
async def get_examples_api():
"""获取计算示例API"""
return {"content": get_calculation_examples()}
@app.get("/api/resources/status")
async def get_status_api():
"""获取计算器状态API"""
return {"content": get_calculator_status()}
# ================== Prompt APIs ==================
@app.post("/api/prompts/math-solve")
async def math_solve_api(request_data: MathSolveRequest):
"""
数学问题解决提示词模板API
生成用于解决数学问题的结构化提示词，包含解题步骤和方法指导。
参数：
- problem: 要解决的数学问题描述
- step_by_step: 是否需要详细的解题步骤（默认True）
返回生成的提示词模板，可用于指导AI解决数学问题。
"""
try:
prompt = math_problem_solver(request_data.problem, request_data.step_by_step)
return {
"prompt_type": "math-solve",
"parameters": {
"problem": request_data.problem,
"step_by_step": request_data.step_by_step
},
"generated_prompt": prompt
}
except Exception as e:
logger.error(f"Math solve prompt API error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/prompts/expression-optimize")
async def expression_optimize_api(request_data: ExpressionOptimizeRequest):
"""
数学表达式优化提示词模板API
生成用于优化数学表达式的提示词，包含优化策略和方法。
参数：
- expression: 需要优化的数学表达式
返回优化指导的提示词模板。
"""
try:
prompt = expression_optimizer(request_data.expression)
return {
"prompt_type": "expression-optimize",
"parameters": {
"expression": request_data.expression
},
"generated_prompt": prompt
}
except Exception as e:
logger.error(f"Expression optimize prompt API error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/prompts/error-debug")
async def error_debug_api(request_data: ErrorDebugRequest):
"""
计算错误调试提示词模板API
生成用于调试计算错误的结构化提示词，包含错误分析步骤。
参数：
- expression: 出错的数学表达式
- error: 错误信息描述
返回错误调试指导的提示词模板。
"""
try:
prompt = calculation_error_debug(request_data.expression, request_data.error)
return {
"prompt_type": "error-debug",
"parameters": {
"expression": request_data.expression,
"error": request_data.error
},
"generated_prompt": prompt
}
except Exception as e:
logger.error(f"Error debug prompt API error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/prompts/math-explain")
async def math_explain_api(request_data: MathExplainRequest):
"""
数学概念解释提示词模板API
生成用于解释数学概念的结构化提示词，根据不同水平提供解释。
参数：
- concept: 要解释的数学概念
- level: 解释水平 (basic/intermediate/advanced)
返回概念解释指导的提示词模板。
"""
try:
prompt = math_concept_explain(request_data.concept, request_data.level)
return {
"prompt_type": "math-explain",
"parameters": {
"concept": request_data.concept,
"level": request_data.level
},
"generated_prompt": prompt
}
except Exception as e:
logger.error(f"Math explain prompt API error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/prompts/formula-derive")
async def formula_derive_api(request_data: FormulaDeriveRequest):
"""
公式推导提示词模板API
生成用于推导数学公式的结构化提示词，包含推导步骤和验证方法。
参数：
- formula_name: 要推导的公式名称
- context: 推导的背景或上下文（可选）
返回公式推导指导的提示词模板。
"""
try:
prompt = formula_derivation(request_data.formula_name, request_data.context)
return {
"prompt_type": "formula-derive",
"parameters": {
"formula_name": request_data.formula_name,
"context": request_data.context
},
"generated_prompt": prompt
}
except Exception as e:
logger.error(f"Formula derive prompt API error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/")
async def root():
"""根路径"""
performance_summary = performance_monitor.get_stats()
return {
"message": "SSE Calculator MCP Server - OAuth鉴权版",
"version": "2.1.0",
"protocol": "MCP over SSE with OAuth 2.0",
"security_features": [
"🔐 OAuth 2.0 鉴权 - 标准授权协议",
"🎫 JWT令牌认证 - 安全的无状态认证",
"👥 用户角色管理 - 支持管理员和普通用户",
"🔑 客户端凭据管理 - 安全的应用程序访问",
"🛡️ 作用域权限控制 - 精细化权限管理",
"🔄 令牌刷新机制 - 无缝续期访问"
],
"optimization_features": [
"⚡ 智能结果缓存 - 提高响应速度",
"🔒 并发请求控制 - 限制最大20个并发任务",
"📊 实时性能监控 - 响应时间和成功率追踪",
"🛡️ 速率限制保护 - 防止服务器过载",
"💾 内存优化管理 - 使用弱引用减少内存占用",
"🔄 异步任务处理 - 非阻塞计算执行"
],
"core_features": [
"数学表达式计算",
"MCP Tools集成",
"MCP Resources文档",
"MCP Prompts模板",
"计算历史记录",
"实时SSE连接",
"OAuth鉴权管理"
],
"oauth_info": {
"authorization_endpoint": "/authorize",
"token_endpoint": "/token",
"refresh_endpoint": "/refresh",
"supported_grant_types": [
"authorization_code",
"client_credentials",
"password",
"refresh_token"
],
"supported_scopes": ["read", "write", "admin"],
"default_credentials": {
"admin_username": "admin",
"admin_password": "admin123",
"client_id": default_client.client_id,
"client_secret": default_client.client_secret[:10] + "..."
}
},
"performance_summary": {
"total_requests": performance_summary.get("total_requests", 0),
"success_rate": f"{performance_summary.get('success_rate', 0)}%",
"cache_hit_rate": f"{performance_summary.get('cache_hit_rate', 0)}%",
"avg_response_time": f"{performance_summary.get('avg_response_time_ms', 0)}ms",
"uptime": f"{performance_summary.get('uptime_seconds', 0)}s"
},
"oauth_endpoints": {
"/token": "获取访问令牌 (POST)",
"/refresh": "刷新访问令牌 (POST)",
"/authorize": "授权端点 (GET)",
"/oauth/info": "OAuth服务器信息 (GET)",
"/oauth/me": "当前用户信息 (GET, 需要认证)",
"/oauth/clients": "客户端管理 (GET/POST, 需要管理员)",
"/oauth/users": "用户管理 (GET/POST, 需要管理员)",
"/oauth/tokens": "令牌管理 (GET, 需要管理员)",
"/oauth/revoke": "撤销令牌 (POST, 需要认证)"
},
"protected_endpoints": {
"/calculate": "计算接口 (需要read或write权限)",
"/trigonometric_functions": "三角函数计算 (需要read或write权限)",
"/logarithmic_functions": "对数函数计算 (需要read或write权限)",
"/statistical_analysis": "统计分析 (需要read或write权限)",
"/polynomial_evaluation": "多项式求值 (需要read或write权限)",
"/complex_number_operations": "复数运算 (需要read或write权限)",
"/expression_parser_advanced": "高级表达式解析 (需要read或write权限)",
"/performance": "性能监控 (需要read或admin权限)"
},
"endpoints": {
"/sse": "SSE连接端点",
"/health": "健康检查 (含性能摘要)",
"/docs": "API文档",
"/api/resources/*": "资源访问API (优化缓存)",
"/api/prompts/*": "提示词模板API"
},
"rate_limits": {
"calculate": "60 requests/minute per IP",
"performance": "30 requests/minute per IP",
"history": "120 requests/minute per IP",
"user_level": "100 requests/minute per IP"
},
"mcp_resources": [
"file://calculator/history",
"file://calculator/functions",
"file://calculator/constants",
"file://calculator/examples",
"file://calculator/status"
],
"mcp_prompts": [
"math-solve",
"expression-optimize",
"error-debug",
"math-explain",
"formula-derive"
],
"authentication_guide": {
"step1": "获取令牌: POST /token with credentials",
"step2": "在请求头中包含: Authorization: Bearer <token>",
"step3": "使用相应权限访问受保护的端点",
"example_curl": "curl -X POST http://localhost:8000/token -d 'username=admin&password=admin123'"
}
}
async def start_sse_server(host: str = "127.0.0.1", port: int = 8000):
"""启动SSE服务器"""
config = uvicorn.Config(
app=app,
host=host,
port=port,
log_level="info",
access_log=True
)
server = uvicorn.Server(config)
logger.info(f"启动SSE Calculator MCP服务器: http://{host}:{port}")
logger.info(f"SSE端点: http://{host}:{port}/sse")
logger.info(f"计算接口: http://{host}:{port}/calculate")
logger.info(f"API文档: http://{host}:{port}/docs")
await server.serve()
async def main():
"""主函数"""
await start_sse_server()
if __name__ == "__main__":
asyncio.run(main())企业级MCP客户端
python
#!/usr/bin/env python3
"""
MCP SSE计算器客户端
连接到SSE计算器服务器，提供计算器功能的客户端接口
"""
import asyncio
import json
import logging
import aiohttp
import time
from typing import Dict, Any, Optional, AsyncGenerator, Callable, List
from contextlib import asynccontextmanager
# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SSECalculatorClient:
"""MCP SSE计算器客户端"""
def __init__(self, server_url: str = "http://127.0.0.1:8000"):
"""
初始化SSE计算器客户端
Args:
server_url: 服务器基础URL
"""
self.server_url = server_url
self.sse_url = f"{server_url}/sse"
self.calculate_url = f"{server_url}/calculate"
self.health_url = f"{server_url}/health"
self.session: Optional[aiohttp.ClientSession] = None
self.is_connected = False
self.server_info: Optional[Dict[str, Any]] = None
# 事件回调
self.on_connection = None
self.on_server_info = None
self.on_heartbeat = None
self.on_error = None
self.on_disconnect = None
async def __aenter__(self):
"""异步上下文管理器入口"""
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口"""
await self.disconnect()
async def connect(self):
"""连接到服务器"""
if self.session is None:
self.session = aiohttp.ClientSession()
logger.info(f"连接到SSE计算器服务器: {self.server_url}")
async def disconnect(self):
"""断开连接"""
if self.session:
await self.session.close()
self.session = None
self.is_connected = False
logger.info("已断开与SSE计算器服务器的连接")
if self.on_disconnect:
await self.on_disconnect()
async def health_check(self) -> Dict[str, Any]:
"""健康检查"""
if not self.session:
await self.connect()
try:
async with self.session.get(self.health_url) as response:
if response.status == 200:
data = await response.json()
logger.info(f"服务器健康检查通过: {data}")
return data
else:
raise Exception(f"健康检查失败: HTTP {response.status}")
except Exception as e:
logger.error(f"健康检查错误: {e}")
raise
async def listen_sse_events(self) -> AsyncGenerator[Dict[str, Any], None]:
"""监听SSE事件流"""
if not self.session:
await self.connect()
try:
logger.info(f"开始监听SSE事件流: {self.sse_url}")
async with self.session.get(self.sse_url) as response:
if response.status != 200:
raise Exception(f"SSE连接失败: HTTP {response.status}")
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data_str = line[6:] # 移除 'data: ' 前缀
try:
event_data = json.loads(data_str)
yield event_data
# 处理不同类型的事件
await self._handle_sse_event(event_data)
except json.JSONDecodeError as e:
logger.warning(f"无法解析SSE数据: {data_str}, 错误: {e}")
except asyncio.CancelledError:
logger.info("SSE事件监听被取消")
except Exception as e:
logger.error(f"SSE事件监听错误: {e}")
if self.on_error:
await self.on_error(str(e))
raise
async def _handle_sse_event(self, event_data: Dict[str, Any]):
"""处理SSE事件"""
event_type = event_data.get('type', '')
if event_type == 'connection':
self.is_connected = True
logger.info(f"SSE连接已建立: {event_data}")
if self.on_connection:
await self.on_connection(event_data)
elif event_type == 'server_info':
self.server_info = event_data.get('data', {})
logger.info(f"收到服务器信息: {self.server_info}")
if self.on_server_info:
await self.on_server_info(self.server_info)
elif event_type == 'heartbeat':
logger.debug(f"收到心跳: {event_data}")
if self.on_heartbeat:
await self.on_heartbeat(event_data)
elif event_type == 'error':
error_msg = event_data.get('message', '未知错误')
logger.error(f"服务器错误: {error_msg}")
if self.on_error:
await self.on_error(error_msg)
async def calculate(self, expression: str) -> Dict[str, Any]:
"""
发送计算请求到服务器
Args:
expression: 数学表达式
Returns:
计算结果字典
"""
if not self.session:
await self.connect()
try:
data = {"expression": expression}
async with self.session.post(
self.calculate_url,
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"计算请求成功: {expression} -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"计算请求失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"计算请求错误: {e}")
return {
"success": False,
"error": str(e),
"expression": expression
}
async def add(self, a: float, b: float) -> Dict[str, Any]:
"""加法运算（通过表达式计算）"""
return await self.calculate(f"{a} + {b}")
async def subtract(self, a: float, b: float) -> Dict[str, Any]:
"""减法运算（通过表达式计算）"""
return await self.calculate(f"{a} - {b}")
async def multiply(self, a: float, b: float) -> Dict[str, Any]:
"""乘法运算（通过表达式计算）"""
return await self.calculate(f"{a} * {b}")
async def divide(self, a: float, b: float) -> Dict[str, Any]:
"""除法运算（通过表达式计算）"""
return await self.calculate(f"{a} / {b}")
async def power(self, base: float, exponent: float) -> Dict[str, Any]:
"""幂运算（通过表达式计算）"""
return await self.calculate(f"{base} ** {exponent}")
async def square_root(self, number: float) -> Dict[str, Any]:
"""平方根运算（通过表达式计算）"""
return await self.calculate(f"sqrt({number})")
# ========== 高级数学函数客户端方法 ==========
async def trigonometric_function(self, function: str, angle: float, unit: str = "radians") -> Dict[str, Any]:
"""
高级三角函数计算
Args:
function: 函数名 (sin, cos, tan, asin, acos, atan, sinh, cosh, tanh, asinh, acosh, atanh)
angle: 角度值
unit: 角度单位 ('radians' 或 'degrees')
"""
if not self.session:
await self.connect()
try:
data = {
"function": function,
"angle": angle,
"unit": unit
}
async with self.session.post(
f"{self.server_url}/trigonometric_functions",
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"三角函数计算成功: {function}({angle}) -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"三角函数计算失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"三角函数计算错误: {e}")
return {
"success": False,
"error": str(e),
"function": function,
"angle": angle
}
async def logarithmic_function(self, function: str, value: float, base: Optional[float] = None) -> Dict[str, Any]:
"""
对数函数计算
Args:
function: 对数类型 ('ln', 'log', 'log10', 'log2', 'logn')
value: 真数
base: 底数 (仅对 'logn' 需要)
"""
if not self.session:
await self.connect()
try:
data = {
"function": function,
"value": value
}
if base is not None:
data["base"] = base
async with self.session.post(
f"{self.server_url}/logarithmic_functions",
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"对数计算成功: {function}({value}) -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"对数计算失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"对数计算错误: {e}")
return {
"success": False,
"error": str(e),
"function": function,
"value": value
}
async def statistical_analysis(self, data: List[float]) -> Dict[str, Any]:
"""
统计分析工具
Args:
data: 数据列表
"""
if not self.session:
await self.connect()
try:
request_data = {"data": data}
async with self.session.post(
f"{self.server_url}/statistical_analysis",
json=request_data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"统计分析成功: {len(data)}个数据点 -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"统计分析失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"统计分析错误: {e}")
return {
"success": False,
"error": str(e),
"data": data
}
async def polynomial_evaluation(self, coefficients: List[float], x: float) -> Dict[str, Any]:
"""
多项式求值
Args:
coefficients: 多项式系数（高次到低次）
x: 自变量值
"""
if not self.session:
await self.connect()
try:
data = {
"coefficients": coefficients,
"x": x
}
async with self.session.post(
f"{self.server_url}/polynomial_evaluation",
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"多项式求值成功: P({x}) -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"多项式求值失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"多项式求值错误: {e}")
return {
"success": False,
"error": str(e),
"coefficients": coefficients,
"x": x
}
async def complex_number_operation(self, operation: str, z1_real: float, z1_imag: float,
z2_real: Optional[float] = None, z2_imag: Optional[float] = None) -> Dict[str, Any]:
"""
复数运算工具
Args:
operation: 运算类型 ('add', 'subtract', 'multiply', 'divide', 'modulus', 'phase', 'conjugate')
z1_real: 第一个复数的实部
z1_imag: 第一个复数的虚部
z2_real: 第二个复数的实部 (二元运算时需要)
z2_imag: 第二个复数的虚部 (二元运算时需要)
"""
if not self.session:
await self.connect()
try:
data = {
"operation": operation,
"z1_real": z1_real,
"z1_imag": z1_imag
}
if z2_real is not None:
data["z2_real"] = z2_real
if z2_imag is not None:
data["z2_imag"] = z2_imag
async with self.session.post(
f"{self.server_url}/complex_number_operations",
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"复数运算成功: {operation} -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"复数运算失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"复数运算错误: {e}")
return {
"success": False,
"error": str(e),
"operation": operation
}
async def expression_parser_advanced(self, expression: str) -> Dict[str, Any]:
"""
高级数学表达式解析器
Args:
expression: 要解析的数学表达式
"""
if not self.session:
await self.connect()
try:
data = {"expression": expression}
async with self.session.post(
f"{self.server_url}/expression_parser_advanced",
json=data,
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
result = await response.json()
logger.info(f"表达式解析成功: {expression} -> {result}")
return result
else:
error_text = await response.text()
raise Exception(f"表达式解析失败: HTTP {response.status}, {error_text}")
except Exception as e:
logger.error(f"表达式解析错误: {e}")
return {
"success": False,
"error": str(e),
"expression": expression
}
# 便捷方法
async def sin_degrees(self, angle: float) -> Dict[str, Any]:
"""正弦函数（角度）"""
return await self.trigonometric_function("sin", angle, "degrees")
async def cos_degrees(self, angle: float) -> Dict[str, Any]:
"""余弦函数（角度）"""
return await self.trigonometric_function("cos", angle, "degrees")
async def tan_degrees(self, angle: float) -> Dict[str, Any]:
"""正切函数（角度）"""
return await self.trigonometric_function("tan", angle, "degrees")
async def natural_log(self, value: float) -> Dict[str, Any]:
"""自然对数"""
return await self.logarithmic_function("ln", value)
async def log_base_10(self, value: float) -> Dict[str, Any]:
"""常用对数（以10为底）"""
return await self.logarithmic_function("log10", value)
async def log_base_2(self, value: float) -> Dict[str, Any]:
"""二进制对数（以2为底）"""
return await self.logarithmic_function("log2", value)
async def complex_add(self, a_real: float, a_imag: float, b_real: float, b_imag: float) -> Dict[str, Any]:
"""复数加法"""
return await self.complex_number_operation("add", a_real, a_imag, b_real, b_imag)
async def complex_multiply(self, a_real: float, a_imag: float, b_real: float, b_imag: float) -> Dict[str, Any]:
"""复数乘法"""
return await self.complex_number_operation("multiply", a_real, a_imag, b_real, b_imag)
async def complex_modulus(self, real: float, imag: float) -> Dict[str, Any]:
"""复数模长"""
return await self.complex_number_operation("modulus", real, imag)
def set_event_handlers(self,
on_connection: Optional[Callable] = None,
on_server_info: Optional[Callable] = None,
on_heartbeat: Optional[Callable] = None,
on_error: Optional[Callable] = None,
on_disconnect: Optional[Callable] = None):
"""设置事件处理回调"""
if on_connection:
self.on_connection = on_connection
if on_server_info:
self.on_server_info = on_server_info
if on_heartbeat:
self.on_heartbeat = on_heartbeat
if on_error:
self.on_error = on_error
if on_disconnect:
self.on_disconnect = on_disconnect
@asynccontextmanager
async def create_sse_calculator_client(server_url: str = "http://127.0.0.1:8000"):
"""
创建SSE计算器客户端的上下文管理器
Args:
server_url: 服务器URL
Yields:
SSECalculatorClient: 客户端实例
"""
client = SSECalculatorClient(server_url)
try:
await client.connect()
yield client
finally:
await client.disconnect()
async def demo_client():
"""客户端演示"""
print("=== SSE计算器客户端演示 ===")
# 事件处理回调
async def on_connection(event_data):
print(f"✅ 已连接到服务器: {event_data}")
async def on_server_info(server_info):
print(f"📋 服务器信息:")
print(f" 名称: {server_info.get('name')}")
print(f" 版本: {server_info.get('version')}")
print(f" 协议: {server_info.get('protocol')}")
print(f" 描述: {server_info.get('description')}")
async def on_heartbeat(event_data):
print(f"💗 心跳: {event_data.get('timestamp')}")
async def on_error(error_msg):
print(f"❌ 错误: {error_msg}")
async def on_disconnect():
print("🔌 连接已断开")
# 使用客户端
async with create_sse_calculator_client() as client:
# 设置事件处理器
client.set_event_handlers(
on_connection=on_connection,
on_server_info=on_server_info,
on_heartbeat=on_heartbeat,
on_error=on_error,
on_disconnect=on_disconnect
)
# 健康检查
try:
health = await client.health_check()
print(f"🩺 健康检查: {health}")
except Exception as e:
print(f"❌ 健康检查失败: {e}")
return
# 启动SSE事件监听（在后台）
sse_task = None
try:
async def sse_listener():
async for event in client.listen_sse_events():
pass # 事件已在_handle_sse_event中处理
sse_task = asyncio.create_task(sse_listener())
# 等待连接建立
await asyncio.sleep(2)
# 执行一些计算
print("\n🧮 开始计算测试:")
# 基本运算
result = await client.calculate("2 + 3")
print(f"计算: 2 + 3 = {result.get('result')} ({'成功' if result.get('success') else '失败'})")
result = await client.add(10, 5)
print(f"加法: 10 + 5 = {result.get('result')} ({'成功' if result.get('success') else '失败'})")
result = await client.divide(20, 4)
print(f"除法: 20 ÷ 4 = {result.get('result')} ({'成功' if result.get('success') else '失败'})")
# 复杂表达式
result = await client.calculate("sin(pi/2) + cos(0)")
print(f"复杂表达式: sin(π/2) + cos(0) = {result.get('result')} ({'成功' if result.get('success') else '失败'})")
# 平方根
result = await client.square_root(16)
print(f"平方根: √16 = {result.get('result')} ({'成功' if result.get('success') else '失败'})")
# 错误处理
result = await client.divide(1, 0)
print(f"除零测试: {result.get('error')} ({'预期错误' if not result.get('success') else '意外成功'})")
print("\n⏱️ 监听3秒钟的心跳...")
await asyncio.sleep(3)
finally:
if sse_task:
sse_task.cancel()
try:
await sse_task
except asyncio.CancelledError:
pass
print("\n✅ 客户端演示完成")
if __name__ == "__main__":
asyncio.run(demo_client())
总结
通过以上章节的学习，我们掌握了 MCP（Model Context Protocol）的核心技术：
- 协议基础：MCP 是连接 AI 助手与数据源的标准化协议，提供统一的接口规范
- 架构设计：采用客户端-服务器模式，支持多种传输方式和安全机制
- 核心原语：Resources、Tools、Prompts、Roots、Sampling 五个基本构建块
- 企业应用：安全认证、监控治理、性能优化等企业级特性
最佳实践总结
- 安全为先：实施多层安全防护，包括认证、授权、加密和审计
- 性能优化：合理使用缓存、连接池和负载均衡提升系统性能
- 监控治理：建立完善的监控体系和治理框架确保系统稳定运行
- 标准遵循：严格遵循 MCP 协议规范，确保兼容性和互操作性