92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""
|
||
MCP Excel 分析服务（FastMCP + Streamable HTTP）。
|
||
|
||
部署建议：Docker 独立容器，backend 通过 MCP_URL 远程调用。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
from mcp.server.fastmcp import FastMCP
|
||
from mcp.server.transport_security import TransportSecuritySettings
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
from starlette.requests import Request
|
||
from starlette.responses import JSONResponse
|
||
|
||
from excel_analyzer import get_excel_schema, run_pandas_on_excel
|
||
|
||
load_dotenv(Path(__file__).resolve().parent / ".env")
|
||
|
||
HOST = os.getenv("MCP_HOST", "0.0.0.0")
|
||
PORT = int(os.getenv("MCP_PORT", "8010"))
|
||
MCP_API_KEY = (os.getenv("MCP_API_KEY") or "").strip()
|
||
|
||
_allowed_hosts = os.getenv("MCP_ALLOWED_HOSTS", f"127.0.0.1:*,localhost:*,[::1]:*")
|
||
_transport_security = TransportSecuritySettings(
|
||
enable_dns_rebinding_protection=os.getenv("MCP_DNS_REBINDING_PROTECTION", "false").lower()
|
||
in ("1", "true", "yes"),
|
||
allowed_hosts=[h.strip() for h in _allowed_hosts.split(",") if h.strip()],
|
||
)
|
||
|
||
mcp = FastMCP(
|
||
"ExcelAnalyzerMCP",
|
||
host=HOST,
|
||
port=PORT,
|
||
stateless_http=True,
|
||
transport_security=_transport_security,
|
||
)
|
||
|
||
|
||
@mcp.tool()
|
||
def excel_get_schema(file_url: str, sheet_name: str = "") -> str:
|
||
"""
|
||
获取 Excel/CSV 的结构信息（sheet 名、列名、类型、前 5 行预览）。
|
||
|
||
在用户提问涉及表格数据前，应先调用本工具了解表格结构，再编写 pandas 代码。
|
||
|
||
Args:
|
||
file_url: 可公网/OSS 内网访问的 Excel/CSV 下载地址（建议 backend 提供带签名的临时 URL）
|
||
sheet_name: 可选，指定 sheet 名或索引（默认第一个 sheet）；CSV 可留空
|
||
"""
|
||
return get_excel_schema(file_url, sheet_name)
|
||
|
||
|
||
@mcp.tool()
|
||
def excel_run_pandas(file_url: str, pandas_code: str, sheet_name: str = "0") -> str:
|
||
"""
|
||
在隔离沙箱中对 Excel/CSV 执行 pandas 分析代码。
|
||
|
||
规则：
|
||
- 仅可使用 ``pd``、``df``；禁止 import 与文件/网络操作
|
||
- 必须将最终答案赋给 ``result``（标量、Series、DataFrame 或字符串均可）
|
||
- 典型流程：先 ``excel_get_schema`` → 再本工具
|
||
|
||
Args:
|
||
file_url: Excel/CSV 下载地址
|
||
pandas_code: pandas 分析代码，例如 ``result = df['销售额'].sum()``
|
||
sheet_name: 要加载的 sheet（名称或索引，默认 0）
|
||
"""
|
||
return run_pandas_on_excel(file_url, pandas_code, sheet_name)
|
||
|
||
|
||
class _APIKeyMiddleware(BaseHTTPMiddleware):
|
||
async def dispatch(self, request: Request, call_next):
|
||
if not MCP_API_KEY:
|
||
return await call_next(request)
|
||
auth = request.headers.get("Authorization", "")
|
||
if auth != f"Bearer {MCP_API_KEY}":
|
||
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
||
return await call_next(request)
|
||
|
||
|
||
app = mcp.streamable_http_app()
|
||
app.add_middleware(_APIKeyMiddleware)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
|
||
uvicorn.run(app, host=HOST, port=PORT)
|