Initial commit: Azure AI Search Agent (Function App)

Made-with: Cursor

.gitignore

@@ -0,0 +1,13 @@
+# 本地配置（含密钥，勿提交）
+local.settings.json
+
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+venv/
+.env
+
+# Azure Functions
+.python_packages/
+.azure/

README.md

@@ -0,0 +1,310 @@
+# Azure AI Search Agent
+
+基于 **Pydantic AI** 的 Azure AI Search 资源库管理 Agent，作为 **OpenClaw 的外部资源库**。
+
+采用 **混合搜索**（关键词 BM25 + 向量语义），支持项目文档的上传、下载、修改、搜索和删除。
+调用方可以是 OpenClaw 也可以是人类用户。
+
+## 搜索模式
+
+| 模式 | 说明 | 适用场景 |
+|------|------|----------|
+| `hybrid`（默认） | 关键词 + 向量融合排序 | 通用场景，推荐 |
+| `keyword` | 纯 BM25 关键词匹配 | 精确搜索术语/ID |
+| `vector` | 纯语义向量搜索 | 自然语言提问，找语义相关文档 |
+
+- 上传文档时自动通过 LiteLLM 生成 embedding 向量（`taiji/text-embedding-3-small`，1536 维）
+- 更新 `title` 或 `content` 时自动重新生成向量
+- 搜索时关键词和向量结果由 Azure AI Search 自动融合排序（RRF）
+
+## 功能
+
+| 工具 | 说明 |
+|------|------|
+| `create_index` | 创建混合搜索索引（含向量字段） |
+| `list_indexes` | 列出所有索引 |
+| `upload_documents` | 上传文档（自动生成向量，支持批量） |
+| `search_documents` | 混合/关键词/向量搜索，支持筛选、排序 |
+| `get_document` | 根据 ID 获取单个文档 |
+| `update_document` | 更新文档部分字段（自动重新生成向量） |
+| `delete_documents` | 删除文档 |
+| `smart_query` | AI 智能问答（混合搜索 + LLM 总结） |
+
+## 文档 Schema
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| `id` | String (Key) | 文档唯一 ID |
+| `title` | String (Searchable) | 标题 |
+| `content` | String (Searchable) | 正文内容 |
+| `content_vector` | Collection(Single) | 内容向量（1536 维，自动生成） |
+| `project` | String (Filterable) | 项目名称 |
+| `category` | String (Facetable) | 分类 |
+| `tags` | String (Filterable) | 标签，逗号分隔 |
+| `source` | String (Filterable) | 来源（openclaw / human） |
+| `author` | String (Filterable) | 作者 |
+| `created_at` | DateTimeOffset | 创建时间 |
+| `updated_at` | DateTimeOffset | 更新时间 |
+| `metadata` | String (Searchable) | 额外元数据 JSON |
+
+## 如何运行（标准 Azure 函数）
+
+在项目根目录 `azure_search_agent/` 下按顺序执行：
+
+```bash
+# 1. 创建并激活虚拟环境（推荐 Python 3.11）
+python3.11 -m venv .venv
+source .venv/bin/activate   # Linux/macOS
+# .venv\Scripts\activate   # Windows
+
+# 2. 安装依赖
+pip install -r requirements.txt
+
+# 3. 确认已有 local.settings.json（你已填写），直接启动
+func start
+```
+
+启动成功后终端会显示：
+
+```
+Functions:
+        http_app: [GET,POST,DELETE,HEAD,PATCH,PUT,OPTIONS] http://localhost:7071/{*route}
+```
+
+在浏览器或 curl 访问：
+
+- 根路径：<http://localhost:7071/>
+- 健康检查：<http://localhost:7071/health>
+- 搜索 API：`curl -X POST http://localhost:7071/api/v1/search -H "Content-Type: application/json" -H "api-key: sk-xxx" -d '{"query":"test","top":5}'`
+
+如需先安装 **Azure Functions Core Tools**（尚未安装时）：
+
+```bash
+# Windows (npm)
+npm i -g azure-functions-core-tools@4
+
+# macOS
+brew tap azure/functions && brew install azure-functions-core-tools@4
+
+# Linux (Ubuntu/Debian)
+curl https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor > microsoft.gpg
+sudo mv microsoft.gpg /etc/apt/trusted.gpg.d/microsoft.gpg
+sudo sh -c 'echo "deb [arch=amd64] https://packages.microsoft.com/repos/microsoft-ubuntu-$(lsb_release -cs)-prod $(lsb_release -cs) main" > /etc/apt/sources.list.d/dotnetdev.list'
+sudo apt-get update && sudo apt-get install azure-functions-core-tools-4
+```
+
+---
+
+## 快速开始
+
+### 1. 配置环境变量
+
+```bash
+export AZURE_SEARCH_ENDPOINT="https://your-search-service.search.windows.net"
+export AZURE_SEARCH_API_KEY="your-admin-key"
+export AZURE_SEARCH_INDEX_NAME="openclaw-resources"
+
+# LiteLLM（用于 embedding 和 smart_query）
+export OPENAI_BASE_URL="https://litellm.example.com/v1"
+export OPENAI_API_KEY="sk-your-key"
+export EMBEDDING_MODEL="taiji/text-embedding-3-small"
+```
+
+### 2. 本地测试
+
+```bash
+pip install -r requirements.txt
+python run_api_server.py
+```
+
+## 部署到 Azure Function App
+
+本 Agent 支持直接部署到 **Azure 函数应用**（无需 AKS）。
+
+### 1. 安装 Azure Functions Core Tools
+
+```bash
+# Windows (npm)
+npm i -g azure-functions-core-tools@4
+
+# macOS (Homebrew)
+brew tap azure/functions && brew install azure-functions-core-tools@4
+
+# Linux 见: https://learn.microsoft.com/azure/azure-functions/functions-run-local
+```
+
+### 2. 本地配置
+
+```bash
+cp local.settings.json.example local.settings.json
+# 编辑 local.settings.json，填入 AZURE_SEARCH_*、OPENAI_* 等
+```
+
+**注意**：`local.settings.json` 含密钥，不要提交到 Git（已在 .gitignore 中忽略）。
+
+### 3. 本地运行（模拟 Function App）
+
+```bash
+pip install -r requirements.txt
+func start
+```
+
+默认地址：`http://localhost:7071`。例如健康检查：`http://localhost:7071/health`。
+
+### 4. 部署到 Azure
+
+**方式 A：Azure CLI**
+
+```bash
+# 登录
+az login
+
+# 创建资源组与存储（若尚未创建）
+az group create --name rg-azure-search-agent --location eastasia
+az storage account create --name styouragent --resource-group rg-azure-search-agent --sku Standard_LRS
+
+# 创建 Function App（Linux + Python 3.11）
+az functionapp create \
+  --resource-group rg-azure-search-agent \
+  --consumption-plan-location eastasia \
+  --runtime python \
+  --runtime-version 3.11 \
+  --functions-version 4 \
+  --name func-azure-search-agent \
+  --storage-account styouragent \
+  --os-type Linux
+
+# 配置应用设置（与 local.settings.json 中的 Values 对应）
+az functionapp config appsettings set --name func-azure-search-agent --resource-group rg-azure-search-agent --settings \
+  AZURE_SEARCH_ENDPOINT="https://aiagnet.search.windows.net" \
+  AZURE_SEARCH_API_KEY="<你的密钥>" \
+  AZURE_SEARCH_INDEX_NAME="openclaw-resources" \
+  OPENAI_BASE_URL="https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1" \
+  OPENAI_API_KEY="<你的密钥>" \
+  EMBEDDING_MODEL="taiji/text-embedding-3-small"
+
+# 部署代码（在项目根目录执行）
+func azure functionapp publish func-azure-search-agent
+```
+
+**方式 B：VS Code**
+
+安装 [Azure Functions 扩展](https://marketplace.visualstudio.com/items?itemName=ms-azuretools.vscode-azurefunctions)，右键项目 → “Deploy to Function App” → 选择或新建 Function App。
+
+### 5. 部署后访问
+
+- 根路径：`https://<your-function-app>.azurewebsites.net/`
+- 健康检查：`https://<your-function-app>.azurewebsites.net/health`
+- REST API：`https://<your-function-app>.azurewebsites.net/api/v1/search` 等
+
+业务接口仍通过请求头 `api-key` 或 `Authorization: Bearer <key>` 鉴权，与本地/uvicorn 行为一致。
+
+### 6. 说明
+
+- **host.json** 中 `routePrefix: ""` 必须保留，否则 FastAPI 路由会多一层前缀。
+- 如需生产级鉴权，可在 Azure 门户将 HTTP 触发器改为 `FUNCTION` 级别，或在前端加 APIM/网关。
+- 消费计划（Consumption）有冷启动与超时限制；长时间运行或高并发可考虑 **Premium** 或 **Dedicated** 计划。
+
+## API 接口
+
+### MCP 端点
+
+- `POST /mcp` — MCP JSON-RPC
+- `GET /mcp/sse` — MCP SSE 连接
+- `POST /mcp/sse` — MCP SSE 请求
+
+### REST API
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| POST | `/api/v1/indexes` | 创建索引 |
+| GET | `/api/v1/indexes` | 列出索引 |
+| POST | `/api/v1/upload` | 上传文档 |
+| POST | `/api/v1/search` | 搜索文档 |
+| GET | `/api/v1/documents/{id}` | 获取文档 |
+| PATCH | `/api/v1/documents/{id}` | 更新文档 |
+| DELETE | `/api/v1/documents/{id}` | 删除文档 |
+| POST | `/api/v1/smart-query` | AI 智能问答 |
+
+### 使用示例
+
+**上传文档：**
+
+```bash
+curl -X POST http://localhost:8000/api/v1/upload \
+  -H "api-key: YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "documents": [
+      {
+        "id": "doc-001",
+        "title": "项目架构设计文档",
+        "content": "本项目采用微服务架构...",
+        "project": "my-project",
+        "category": "设计文档",
+        "source": "human",
+        "author": "张三"
+      }
+    ]
+  }'
+```
+
+**混合搜索（默认）：**
+
+```bash
+curl -X POST http://localhost:8000/api/v1/search \
+  -H "api-key: YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"query": "系统用了什么技术栈", "top": 5}'
+```
+
+**纯关键词搜索：**
+
+```bash
+curl -X POST http://localhost:8000/api/v1/search \
+  -H "api-key: YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"query": "微服务", "search_mode": "keyword"}'
+```
+
+**AI 智能问答：**
+
+```bash
+curl -X POST http://localhost:8000/api/v1/smart-query \
+  -H "api-key: YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"question": "怎么开发新的 Agent？", "project": "openclaw"}'
+```
+
+## 环境变量
+
+| 变量 | 必需 | 说明 |
+|------|------|------|
+| AZURE_SEARCH_ENDPOINT | 是 | Azure AI Search 服务端点 URL |
+| AZURE_SEARCH_API_KEY | 是 | Azure AI Search 管理密钥 |
+| AZURE_SEARCH_INDEX_NAME | 否 | 默认索引名称，默认 `openclaw-resources` |
+| OPENAI_BASE_URL | 否 | LiteLLM Gateway URL（embedding + 智能问答） |
+| OPENAI_API_KEY | 否 | LiteLLM API Key |
+| EMBEDDING_MODEL | 否 | Embedding 模型名称，默认 `taiji/text-embedding-3-small` |
+| EMBEDDING_DIMENSIONS | 否 | 向量维度，默认 `1536` |
+| LITELLM_MODEL | 否 | LLM 模型名称，默认 `taiji/gpt-4o-mini` |
+| API_PORT | 否 | 端口，默认 8000 |
+
+## 项目结构
+
+```
+azure_search_agent/
+├── README.md
+├── requirements.txt
+├── run_api_server.py        # 本地 uvicorn 启动（非 Function App）
+├── function_app.py         # Azure Function App 入口（ASGI 挂载 FastAPI）
+├── host.json               # Function 主机配置（routePrefix 为空）
+├── local.settings.json.example
+├── local.settings.json     # 本地配置（勿提交，复制 example 后填写）
+└── src/
+    ├── __init__.py
+    └── server/
+        ├── __init__.py
+        ├── api_server.py   # FastAPI + MCP HTTP + REST API
+        └── mcp_server.py   # MCP 工具定义（混合搜索 + embedding）
+```

function_app.py

@@ -0,0 +1,14 @@
+"""
+Azure Function App 入口
+
+将 FastAPI 应用挂载为 ASGI，用于部署到 Azure 函数应用（非 AKS）。
+本地运行: func start
+部署后: https://<your-function-app>.azurewebsites.net/
+"""
+import azure.functions as func
+
+from src.server.api_server import app as fastapi_app
+
+# 挂载 FastAPI 为 ASGI，所有 HTTP 请求由 FastAPI 处理
+# ANONYMOUS: 允许匿名调用；生产环境可在 Azure 门户改为 FUNCTION 并配合密钥/APIM
+app = func.AsgiFunctionApp(app=fastapi_app, http_auth_level=func.AuthLevel.ANONYMOUS)

host.json

@@ -0,0 +1,13 @@
+{
+  "version": "2.0",
+  "extensionBundle": {
+    "id": "Microsoft.Azure.Functions.ExtensionBundle",
+    "version": "[4.*, 5.0.0)"
+  },
+  "extensions": {
+    "http": {
+      "routePrefix": ""
+    }
+  },
+  "functionTimeout": "00:10:00"
+}

local.settings.json.example

@@ -0,0 +1,15 @@
+{
+  "IsEncrypted": false,
+  "Values": {
+    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
+    "FUNCTIONS_WORKER_RUNTIME": "python",
+
+    "AZURE_SEARCH_ENDPOINT": "https://<your-search>.search.windows.net",
+    "AZURE_SEARCH_API_KEY": "<your-search-admin-key>",
+    "AZURE_SEARCH_INDEX_NAME": "openclaw-resources",
+
+    "OPENAI_BASE_URL": "https://litellm.example.com/v1",
+    "OPENAI_API_KEY": "sk-your-key",
+    "EMBEDDING_MODEL": "taiji/text-embedding-3-small"
+  }
+}

requirements.txt

@@ -0,0 +1,20 @@
+# Azure Functions（部署到 Function App 时需要）
+azure-functions>=1.17.0
+
+# Pydantic AI
+pydantic-ai>=0.0.14
+
+# MCP
+mcp>=0.9.0
+fastmcp>=0.1.0
+
+# FastAPI
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+
+# HTTP Client
+aiohttp>=3.9.0
+
+# Azure AI Search
+azure-search-documents>=11.6.0
+azure-core>=1.30.0

run_api_server.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python
+"""启动 Azure AI Search Agent API 服务器"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+if __name__ == "__main__":
+    from src.server.api_server import app
+    import uvicorn
+    import os
+
+    host = os.getenv("API_HOST", "0.0.0.0")
+    port = int(os.getenv("API_PORT", "8000"))
+
+    print(f"🚀 启动 Azure AI Search Agent API: http://{host}:{port}")
+    uvicorn.run(app, host=host, port=port, log_level="info")

src/__init__.py

@@ -0,0 +1 @@
+"""Azure AI Search Agent 源代码包"""

src/server/__init__.py

@@ -0,0 +1 @@
+"""服务器模块"""

src/server/api_server.py

@@ -0,0 +1,363 @@
+"""
+Azure AI Search Agent - HTTP API 服务器
+
+提供 REST API 和 MCP HTTP/SSE 端点。
+支持 OpenClaw 和人类用户通过 HTTP 接口操作资源库。
+"""
+import json
+import uuid
+import os
+from typing import Optional, Dict, Any, List, AsyncGenerator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException, Request, Header, Depends
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse, JSONResponse
+from pydantic import BaseModel, Field
+
+from .mcp_server import TOOL_MAP, TOOL_LIST
+
+# ==================== 配置 ====================
+
+SERVER_NAME = "Azure AI Search Agent API"
+
+
+# ==================== FastAPI 应用 ====================
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print(f"🚀 {SERVER_NAME} 启动")
+    yield
+    print(f"🛑 {SERVER_NAME} 关闭")
+
+
+app = FastAPI(
+    title=SERVER_NAME,
+    description="OpenClaw 外部资源库 - 基于 Azure AI Search 的文档管理与搜索服务",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ==================== API Key 验证 ====================
+
+async def verify_api_key(
+    api_key: Optional[str] = Header(None, alias="api-key"),
+    authorization: Optional[str] = Header(None),
+) -> str:
+    if api_key and api_key.strip() and api_key.strip() != "sk":
+        return api_key.strip()
+    if authorization:
+        key = authorization[7:].strip() if authorization.startswith("Bearer ") else authorization.strip()
+        if key and key != "sk":
+            return key
+    raise HTTPException(status_code=401, detail="缺少 API Key")
+
+
+def get_api_key_from_request(request: Request) -> Optional[str]:
+    api_key = request.headers.get("api-key") or request.headers.get("api_key")
+    if not api_key:
+        auth = request.headers.get("Authorization")
+        if auth:
+            api_key = auth[7:] if auth.startswith("Bearer ") else auth
+    return api_key
+
+
+# ==================== 健康检查 ====================
+
+@app.get("/")
+async def root():
+    return {
+        "service": SERVER_NAME,
+        "status": "running",
+        "tools": list(TOOL_MAP.keys()),
+    }
+
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy", "service": SERVER_NAME}
+
+
+# ==================== MCP 端点 ====================
+
+sessions: Dict[str, Dict] = {}
+
+
+async def handle_mcp_request(data: Dict, session_id: str = None, api_key: str = None) -> Dict:
+    method = data.get("method")
+    params = data.get("params", {})
+    req_id = data.get("id")
+
+    if method == "tools/call" and (not api_key or api_key == "sk"):
+        return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32001, "message": "缺少 API Key"}}
+
+    try:
+        if method == "initialize":
+            session_id = session_id or str(uuid.uuid4())
+            sessions[session_id] = {"initialized": True}
+            return {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "result": {
+                    "protocolVersion": "2024-11-05",
+                    "capabilities": {"tools": {}},
+                    "serverInfo": {"name": SERVER_NAME, "version": "1.0.0"},
+                },
+            }
+        elif method == "tools/list":
+            return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOL_LIST}}
+        elif method == "tools/call":
+            tool_name = params.get("name")
+            args = params.get("arguments", {})
+            if tool_name not in TOOL_MAP:
+                raise ValueError(f"Unknown tool: {tool_name}")
+
+            old_key = os.environ.get("OPENAI_API_KEY")
+            if api_key:
+                os.environ["OPENAI_API_KEY"] = api_key
+            try:
+                result = await TOOL_MAP[tool_name](**args)
+            finally:
+                if old_key:
+                    os.environ["OPENAI_API_KEY"] = old_key
+
+            return {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "result": {"content": [{"type": "text", "text": str(result)}]},
+            }
+        elif method == "ping":
+            return {"jsonrpc": "2.0", "id": req_id, "result": {}}
+        else:
+            raise ValueError(f"Unknown method: {method}")
+    except Exception as e:
+        return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(e)}}
+
+
+@app.post("/mcp")
+async def mcp_endpoint(request: Request):
+    try:
+        body = await request.json()
+        session_id = request.headers.get("x-mcp-session-id")
+        api_key = get_api_key_from_request(request)
+        response = await handle_mcp_request(body, session_id, api_key)
+        return JSONResponse(content=response, headers={"x-mcp-session-id": session_id or ""})
+    except Exception as e:
+        return JSONResponse(
+            status_code=400,
+            content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}},
+        )
+
+
+@app.get("/mcp/sse")
+async def mcp_sse(request: Request):
+    session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
+
+    async def stream() -> AsyncGenerator[str, None]:
+        yield f"data: {json.dumps({'type': 'connection', 'sessionId': session_id})}\n\n"
+        import asyncio
+
+        while True:
+            await asyncio.sleep(30)
+            yield f"data: {json.dumps({'type': 'ping'})}\n\n"
+
+    return StreamingResponse(
+        stream(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id},
+    )
+
+
+@app.post("/mcp/sse")
+async def mcp_sse_post(request: Request):
+    try:
+        body = await request.json()
+        session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
+        api_key = get_api_key_from_request(request)
+
+        async def stream() -> AsyncGenerator[str, None]:
+            response = await handle_mcp_request(body, session_id, api_key)
+            yield f"data: {json.dumps(response)}\n\n"
+
+        return StreamingResponse(
+            stream(),
+            media_type="text/event-stream",
+            headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id},
+        )
+    except Exception as e:
+        return JSONResponse(
+            status_code=400,
+            content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}},
+        )
+
+
+# ==================== 业务 REST API ====================
+
+
+class UploadRequest(BaseModel):
+    documents: List[Dict[str, Any]] = Field(..., description="文档列表")
+    index_name: Optional[str] = Field(None, description="索引名称")
+
+
+class SearchRequest(BaseModel):
+    query: str = Field(..., description="搜索关键词或自然语言查询")
+    index_name: Optional[str] = Field(None, description="索引名称")
+    filter_expr: Optional[str] = Field(None, description="OData 筛选表达式")
+    top: int = Field(10, description="返回数量")
+    select: Optional[str] = Field(None, description="返回字段，逗号分隔")
+    order_by: Optional[str] = Field(None, description="排序字段")
+    search_mode: str = Field("hybrid", description="搜索模式: hybrid / keyword / vector")
+
+
+class UpdateRequest(BaseModel):
+    updates: Dict[str, Any] = Field(..., description="要更新的字段")
+    index_name: Optional[str] = Field(None, description="索引名称")
+
+
+class SmartQueryRequest(BaseModel):
+    question: str = Field(..., description="自然语言问题")
+    index_name: Optional[str] = Field(None, description="索引名称")
+    project: Optional[str] = Field(None, description="项目名称")
+    top: int = Field(5, description="返回结果数量")
+
+
+@app.post("/api/v1/upload")
+async def api_upload(request: UploadRequest, api_key: str = Depends(verify_api_key)):
+    """上传文档到资源库"""
+    try:
+        old_key = os.environ.get("OPENAI_API_KEY")
+        os.environ["OPENAI_API_KEY"] = api_key
+        try:
+            result = await TOOL_MAP["upload_documents"](
+                documents=json.dumps(request.documents), index_name=request.index_name
+            )
+            return json.loads(result)
+        finally:
+            if old_key:
+                os.environ["OPENAI_API_KEY"] = old_key
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/v1/search")
+async def api_search(request: SearchRequest, api_key: str = Depends(verify_api_key)):
+    """搜索资源库"""
+    try:
+        result = await TOOL_MAP["search_documents"](
+            query=request.query,
+            index_name=request.index_name,
+            filter_expr=request.filter_expr,
+            top=request.top,
+            search_mode=request.search_mode,
+            select=request.select,
+            order_by=request.order_by,
+        )
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/v1/documents/{document_id}")
+async def api_get_document(
+    document_id: str,
+    index_name: Optional[str] = None,
+    api_key: str = Depends(verify_api_key),
+):
+    """获取单个文档"""
+    try:
+        result = await TOOL_MAP["get_document"](document_id=document_id, index_name=index_name)
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.patch("/api/v1/documents/{document_id}")
+async def api_update_document(
+    document_id: str,
+    request: UpdateRequest,
+    api_key: str = Depends(verify_api_key),
+):
+    """更新文档"""
+    try:
+        result = await TOOL_MAP["update_document"](
+            document_id=document_id,
+            updates=json.dumps(request.updates),
+            index_name=request.index_name,
+        )
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.delete("/api/v1/documents/{document_id}")
+async def api_delete_document(
+    document_id: str,
+    index_name: Optional[str] = None,
+    api_key: str = Depends(verify_api_key),
+):
+    """删除文档"""
+    try:
+        result = await TOOL_MAP["delete_documents"](document_ids=document_id, index_name=index_name)
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/v1/smart-query")
+async def api_smart_query(request: SmartQueryRequest, api_key: str = Depends(verify_api_key)):
+    """AI 智能问答"""
+    try:
+        old_key = os.environ.get("OPENAI_API_KEY")
+        os.environ["OPENAI_API_KEY"] = api_key
+        try:
+            result = await TOOL_MAP["smart_query"](
+                question=request.question,
+                index_name=request.index_name,
+                project=request.project,
+                top=request.top,
+            )
+            return json.loads(result)
+        finally:
+            if old_key:
+                os.environ["OPENAI_API_KEY"] = old_key
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/v1/indexes")
+async def api_create_index(
+    index_name: str,
+    use_default_schema: bool = True,
+    api_key: str = Depends(verify_api_key),
+):
+    """创建搜索索引"""
+    try:
+        result = await TOOL_MAP["create_index"](index_name=index_name, use_default_schema=use_default_schema)
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/v1/indexes")
+async def api_list_indexes(api_key: str = Depends(verify_api_key)):
+    """列出所有索引"""
+    try:
+        result = await TOOL_MAP["list_indexes"]()
+        return json.loads(result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)

src/server/mcp_server.py

@@ -0,0 +1,720 @@
+"""
+Azure AI Search MCP 服务器（混合搜索版）
+
+为 OpenClaw 提供外部资源库功能，支持项目文档的上传、下载、修改、搜索和删除。
+搜索方式：关键词 (BM25) + 向量 (Embedding) 混合搜索，自动融合排序。
+"""
+import json
+import os
+import logging
+from typing import Optional, List
+from datetime import datetime, timezone
+
+import aiohttp
+from mcp.server.fastmcp import FastMCP
+from pydantic_ai import Agent
+from azure.core.credentials import AzureKeyCredential
+from azure.search.documents import SearchClient, IndexDocumentsBatch
+from azure.search.documents.indexes import SearchIndexClient
+from azure.search.documents.indexes.models import (
+    SearchIndex,
+    SimpleField,
+    SearchableField,
+    SearchField,
+    SearchFieldDataType,
+    CorsOptions,
+    VectorSearch,
+    HnswAlgorithmConfiguration,
+    VectorSearchProfile,
+)
+from azure.search.documents.models import VectorizedQuery
+
+logger = logging.getLogger(__name__)
+
+# ==================== 配置 ====================
+
+AZURE_SEARCH_ENDPOINT = os.getenv("AZURE_SEARCH_ENDPOINT", "")
+AZURE_SEARCH_API_KEY = os.getenv("AZURE_SEARCH_API_KEY", "")
+AZURE_SEARCH_INDEX_NAME = os.getenv("AZURE_SEARCH_INDEX_NAME", "openclaw-resources")
+
+_BASE_URL = os.getenv(
+    "OPENAI_BASE_URL",
+    os.getenv(
+        "LLM_BASE_URL",
+        "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1",
+    ),
+)
+_API_KEY = os.getenv("OPENAI_API_KEY", "sk")
+
+os.environ.setdefault("OPENAI_API_KEY", _API_KEY)
+os.environ.setdefault("OPENAI_BASE_URL", _BASE_URL)
+
+EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "taiji/text-embedding-3-small")
+EMBEDDING_DIMENSIONS = int(os.getenv("EMBEDDING_DIMENSIONS", "1536"))
+
+VECTOR_SEARCH_PROFILE = "default-vector-profile"
+VECTOR_ALGORITHM = "default-hnsw"
+
+
+def _get_model_name() -> str:
+    model = os.getenv("MODEL_NAME", os.getenv("LITELLM_MODEL", "taiji/gpt-4o-mini"))
+    return model if ":" in model else f"openai:{model}"
+
+
+MODEL_NAME = _get_model_name()
+
+# ==================== MCP 服务器 ====================
+
+server = FastMCP("Azure AI Search Agent")
+
+SYSTEM_PROMPT = """你是 OpenClaw 的外部资源库管理助手。
+你可以帮助用户管理 Azure AI Search 中的项目资源文档，包括上传、搜索、下载、修改和删除文档。
+搜索采用混合模式（关键词 + 语义向量），能同时处理精确匹配和语义理解。"""
+
+
+def _build_index_fields() -> list:
+    """构建包含向量字段的索引 Schema"""
+    return [
+        SimpleField(name="id", type=SearchFieldDataType.String, key=True, filterable=True),
+        SearchableField(name="title", type=SearchFieldDataType.String, filterable=True, sortable=True),
+        SearchableField(name="content", type=SearchFieldDataType.String),
+        SimpleField(name="project", type=SearchFieldDataType.String, filterable=True, sortable=True),
+        SearchableField(name="category", type=SearchFieldDataType.String, filterable=True, facetable=True),
+        SearchableField(name="tags", type=SearchFieldDataType.String, filterable=True),
+        SimpleField(name="source", type=SearchFieldDataType.String, filterable=True),
+        SimpleField(name="author", type=SearchFieldDataType.String, filterable=True),
+        SimpleField(name="created_at", type=SearchFieldDataType.DateTimeOffset, filterable=True, sortable=True),
+        SimpleField(name="updated_at", type=SearchFieldDataType.DateTimeOffset, filterable=True, sortable=True),
+        SearchableField(name="metadata", type=SearchFieldDataType.String),
+        SearchField(
+            name="content_vector",
+            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
+            searchable=True,
+            vector_search_dimensions=EMBEDDING_DIMENSIONS,
+            vector_search_profile_name=VECTOR_SEARCH_PROFILE,
+        ),
+    ]
+
+
+def _build_vector_search() -> VectorSearch:
+    """构建向量搜索配置"""
+    return VectorSearch(
+        algorithms=[HnswAlgorithmConfiguration(name=VECTOR_ALGORITHM)],
+        profiles=[
+            VectorSearchProfile(name=VECTOR_SEARCH_PROFILE, algorithm_configuration_name=VECTOR_ALGORITHM),
+        ],
+    )
+
+
+# ==================== Embedding 工具 ====================
+
+
+async def _get_embedding(text: str) -> List[float]:
+    """通过 LiteLLM Gateway 获取文本的 embedding 向量"""
+    url = _BASE_URL.rstrip("/") + "/embeddings"
+    api_key = os.environ.get("OPENAI_API_KEY", _API_KEY)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {"model": EMBEDDING_MODEL, "input": text}
+
+    async with aiohttp.ClientSession() as session:
+        async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=30)) as resp:
+            data = await resp.json()
+            if "data" not in data:
+                err = data.get("error", {}).get("message", str(data))
+                raise ValueError(f"Embedding 请求失败: {err}")
+            return data["data"][0]["embedding"]
+
+
+async def _get_embeddings_batch(texts: List[str]) -> List[List[float]]:
+    """批量获取 embeddings（单次请求）"""
+    url = _BASE_URL.rstrip("/") + "/embeddings"
+    api_key = os.environ.get("OPENAI_API_KEY", _API_KEY)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {"model": EMBEDDING_MODEL, "input": texts}
+
+    async with aiohttp.ClientSession() as session:
+        async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=60)) as resp:
+            data = await resp.json()
+            if "data" not in data:
+                err = data.get("error", {}).get("message", str(data))
+                raise ValueError(f"Embedding 批量请求失败: {err}")
+            sorted_data = sorted(data["data"], key=lambda x: x["index"])
+            return [item["embedding"] for item in sorted_data]
+
+
+def _build_embedding_text(doc: dict) -> str:
+    """拼接文档中用于生成向量的文本"""
+    parts = []
+    if doc.get("title"):
+        parts.append(doc["title"])
+    if doc.get("content"):
+        parts.append(doc["content"][:8000])
+    return "\n".join(parts) if parts else ""
+
+
+# ==================== Azure Search 客户端 ====================
+
+
+def _get_credential() -> AzureKeyCredential:
+    key = os.environ.get("AZURE_SEARCH_API_KEY", AZURE_SEARCH_API_KEY)
+    if not key:
+        raise ValueError("AZURE_SEARCH_API_KEY 未配置")
+    return AzureKeyCredential(key)
+
+
+def _get_endpoint() -> str:
+    endpoint = os.environ.get("AZURE_SEARCH_ENDPOINT", AZURE_SEARCH_ENDPOINT)
+    if not endpoint:
+        raise ValueError("AZURE_SEARCH_ENDPOINT 未配置")
+    return endpoint
+
+
+def _get_index_client() -> SearchIndexClient:
+    return SearchIndexClient(endpoint=_get_endpoint(), credential=_get_credential())
+
+
+def _get_search_client(index_name: Optional[str] = None) -> SearchClient:
+    name = index_name or os.environ.get("AZURE_SEARCH_INDEX_NAME", AZURE_SEARCH_INDEX_NAME)
+    return SearchClient(endpoint=_get_endpoint(), index_name=name, credential=_get_credential())
+
+
+def get_agent() -> Agent:
+    return Agent(MODEL_NAME, system_prompt=SYSTEM_PROMPT)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ==================== MCP 工具定义 ====================
+
+
+@server.tool()
+async def create_index(
+    index_name: str,
+) -> str:
+    """
+    创建支持混合搜索（关键词 + 向量）的 Azure AI Search 索引。首次使用时需先创建索引。
+
+    Args:
+        index_name: 索引名称（如 openclaw-resources）
+
+    Returns:
+        创建结果
+    """
+    try:
+        client = _get_index_client()
+        fields = _build_index_fields()
+        vector_search = _build_vector_search()
+        cors = CorsOptions(allowed_origins=["*"], max_age_in_seconds=60)
+        index = SearchIndex(
+            name=index_name,
+            fields=fields,
+            vector_search=vector_search,
+            cors_options=cors,
+        )
+        result = client.create_or_update_index(index)
+        return json.dumps(
+            {
+                "success": True,
+                "index_name": result.name,
+                "fields_count": len(result.fields),
+                "vector_search": True,
+                "message": f"混合搜索索引 '{result.name}' 创建/更新成功",
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    except Exception as e:
+        logger.error(f"创建索引失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def list_indexes() -> str:
+    """
+    列出所有可用的搜索索引。
+
+    Returns:
+        索引列表
+    """
+    try:
+        client = _get_index_client()
+        indexes = []
+        for idx in client.list_indexes():
+            has_vector = any(
+                getattr(f, "vector_search_dimensions", None) is not None
+                for f in (idx.fields or [])
+            )
+            indexes.append({
+                "name": idx.name,
+                "fields_count": len(idx.fields) if idx.fields else 0,
+                "vector_search": has_vector,
+            })
+        return json.dumps({"success": True, "indexes": indexes}, ensure_ascii=False, indent=2)
+    except Exception as e:
+        logger.error(f"列出索引失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def upload_documents(
+    documents: str,
+    index_name: Optional[str] = None,
+) -> str:
+    """
+    上传文档到 Azure AI Search 索引。自动生成向量用于混合搜索。支持批量上传。
+
+    Args:
+        documents: JSON 格式的文档数组字符串，每个文档应包含:
+            - id: 文档唯一ID（必需）
+            - title: 标题（必需）
+            - content: 正文内容（必需）
+            - project: 项目名称（可选）
+            - category: 分类（可选）
+            - tags: 标签，逗号分隔（可选）
+            - source: 来源，如 "openclaw" 或 "human"（可选）
+            - author: 作者（可选）
+            - metadata: 额外元数据 JSON 字符串（可选）
+        index_name: 索引名称，默认使用环境变量配置
+
+    Returns:
+        上传结果
+    """
+    try:
+        docs = json.loads(documents) if isinstance(documents, str) else documents
+        if isinstance(docs, dict):
+            docs = [docs]
+
+        now = _now_iso()
+        for doc in docs:
+            doc.setdefault("created_at", now)
+            doc.setdefault("updated_at", now)
+            doc.setdefault("source", "unknown")
+
+        texts = [_build_embedding_text(doc) for doc in docs]
+        embeddings = await _get_embeddings_batch(texts)
+        for doc, emb in zip(docs, embeddings):
+            doc["content_vector"] = emb
+
+        client = _get_search_client(index_name)
+        batch = IndexDocumentsBatch()
+        batch.add_upload_actions(docs)
+        results = client.index_documents(batch)
+
+        succeeded = sum(1 for r in results if r.succeeded)
+        failed = sum(1 for r in results if not r.succeeded)
+        errors = [
+            {"key": r.key, "error": r.error_message}
+            for r in results
+            if not r.succeeded
+        ]
+
+        return json.dumps(
+            {
+                "success": failed == 0,
+                "uploaded": succeeded,
+                "failed": failed,
+                "errors": errors,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    except Exception as e:
+        logger.error(f"上传文档失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def search_documents(
+    query: str,
+    index_name: Optional[str] = None,
+    filter_expr: Optional[str] = None,
+    top: int = 10,
+    select: Optional[str] = None,
+    order_by: Optional[str] = None,
+    search_mode: str = "hybrid",
+) -> str:
+    """
+    混合搜索文档（关键词 + 语义向量）。Azure AI Search 自动融合两种结果排序。
+
+    Args:
+        query: 搜索关键词或自然语言查询
+        index_name: 索引名称，默认使用环境变量配置
+        filter_expr: OData 筛选表达式（如 "project eq 'my-project'" 或 "source eq 'openclaw'"）
+        top: 返回结果数量，默认 10
+        select: 返回字段列表，逗号分隔（如 "id,title,content"），默认返回所有字段
+        order_by: 排序字段（如 "updated_at desc"），设置后会覆盖混合排序
+        search_mode: 搜索模式 - "hybrid"（混合，默认）、"keyword"（纯关键词）、"vector"（纯向量）
+
+    Returns:
+        搜索结果列表
+    """
+    try:
+        client = _get_search_client(index_name)
+
+        default_exclude = {"content_vector"}
+        if select:
+            select_fields = [s.strip() for s in select.split(",")]
+        else:
+            select_fields = None
+
+        search_kwargs = {
+            "filter": filter_expr,
+            "top": top,
+            "select": select_fields,
+            "order_by": order_by,
+        }
+
+        if search_mode == "keyword":
+            search_kwargs["search_text"] = query
+        elif search_mode == "vector":
+            query_vector = await _get_embedding(query)
+            search_kwargs["search_text"] = None
+            search_kwargs["vector_queries"] = [
+                VectorizedQuery(vector=query_vector, k_nearest_neighbors=top, fields="content_vector")
+            ]
+        else:
+            query_vector = await _get_embedding(query)
+            search_kwargs["search_text"] = query
+            search_kwargs["vector_queries"] = [
+                VectorizedQuery(vector=query_vector, k_nearest_neighbors=top, fields="content_vector")
+            ]
+
+        results = client.search(**search_kwargs)
+
+        docs = []
+        for result in results:
+            doc = {k: v for k, v in result.items() if not k.startswith("@") and k not in default_exclude}
+            doc["_score"] = result.get("@search.score")
+            docs.append(doc)
+
+        return json.dumps(
+            {"success": True, "count": len(docs), "search_mode": search_mode, "results": docs},
+            ensure_ascii=False,
+            indent=2,
+            default=str,
+        )
+    except Exception as e:
+        logger.error(f"搜索文档失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def get_document(
+    document_id: str,
+    index_name: Optional[str] = None,
+    select: Optional[str] = None,
+) -> str:
+    """
+    根据 ID 获取单个文档（下载）。
+
+    Args:
+        document_id: 文档 ID
+        index_name: 索引名称，默认使用环境变量配置
+        select: 返回字段列表，逗号分隔（可选）
+
+    Returns:
+        文档内容
+    """
+    try:
+        client = _get_search_client(index_name)
+        select_fields = [s.strip() for s in select.split(",")] if select else None
+        doc = client.get_document(key=document_id, selected_fields=select_fields)
+        doc_dict = dict(doc)
+        doc_dict.pop("content_vector", None)
+        return json.dumps(
+            {"success": True, "document": doc_dict},
+            ensure_ascii=False,
+            indent=2,
+            default=str,
+        )
+    except Exception as e:
+        logger.error(f"获取文档失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def update_document(
+    document_id: str,
+    updates: str,
+    index_name: Optional[str] = None,
+) -> str:
+    """
+    更新已有文档的部分字段（merge 方式）。如果更新了 title 或 content，会自动重新生成向量。
+
+    Args:
+        document_id: 文档 ID
+        updates: JSON 格式的更新字段（如 '{"title": "新标题", "content": "新内容"}'）
+        index_name: 索引名称，默认使用环境变量配置
+
+    Returns:
+        更新结果
+    """
+    try:
+        fields = json.loads(updates) if isinstance(updates, str) else updates
+        fields["id"] = document_id
+        fields["updated_at"] = _now_iso()
+
+        if "title" in fields or "content" in fields:
+            client = _get_search_client(index_name)
+            existing = client.get_document(key=document_id)
+            title = fields.get("title", existing.get("title", ""))
+            content = fields.get("content", existing.get("content", ""))
+            emb_text = f"{title}\n{content}" if content else title
+            fields["content_vector"] = await _get_embedding(emb_text)
+
+        client = _get_search_client(index_name)
+        batch = IndexDocumentsBatch()
+        batch.add_merge_actions([fields])
+        results = client.index_documents(batch)
+
+        r = results[0]
+        return json.dumps(
+            {
+                "success": r.succeeded,
+                "document_id": document_id,
+                "vector_updated": "content_vector" in fields,
+                "message": "文档更新成功" if r.succeeded else r.error_message,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    except Exception as e:
+        logger.error(f"更新文档失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def delete_documents(
+    document_ids: str,
+    index_name: Optional[str] = None,
+) -> str:
+    """
+    根据 ID 删除一个或多个文档。
+
+    Args:
+        document_ids: 文档 ID 列表的 JSON 数组字符串（如 '["id1", "id2"]'）或单个 ID 字符串
+        index_name: 索引名称，默认使用环境变量配置
+
+    Returns:
+        删除结果
+    """
+    try:
+        ids = json.loads(document_ids) if document_ids.startswith("[") else [document_ids]
+        docs_to_delete = [{"id": doc_id} for doc_id in ids]
+
+        client = _get_search_client(index_name)
+        batch = IndexDocumentsBatch()
+        batch.add_delete_actions(docs_to_delete)
+        results = client.index_documents(batch)
+
+        succeeded = sum(1 for r in results if r.succeeded)
+        failed = sum(1 for r in results if not r.succeeded)
+
+        return json.dumps(
+            {"success": failed == 0, "deleted": succeeded, "failed": failed},
+            ensure_ascii=False,
+            indent=2,
+        )
+    except Exception as e:
+        logger.error(f"删除文档失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+@server.tool()
+async def smart_query(
+    question: str,
+    index_name: Optional[str] = None,
+    project: Optional[str] = None,
+    top: int = 5,
+) -> str:
+    """
+    智能查询：混合搜索 + AI 总结。使用向量+关键词找到最相关文档，再由 LLM 总结回答。
+    适合 OpenClaw 或人类用户用自然语言提问。
+
+    Args:
+        question: 自然语言问题（如"项目X的最新设计文档是什么？"）
+        index_name: 索引名称，默认使用环境变量配置
+        project: 限定项目名称（可选）
+        top: 返回结果数量
+
+    Returns:
+        AI 总结的回答及原始文档来源
+    """
+    try:
+        filter_expr = f"project eq '{project}'" if project else None
+        client = _get_search_client(index_name)
+
+        query_vector = await _get_embedding(question)
+        results = client.search(
+            search_text=question,
+            vector_queries=[
+                VectorizedQuery(vector=query_vector, k_nearest_neighbors=top, fields="content_vector")
+            ],
+            filter=filter_expr,
+            top=top,
+        )
+
+        docs = []
+        for result in results:
+            docs.append({
+                "id": result.get("id"),
+                "title": result.get("title"),
+                "content": result.get("content", "")[:2000],
+                "project": result.get("project"),
+                "score": result.get("@search.score"),
+            })
+
+        if not docs:
+            return json.dumps(
+                {"success": True, "answer": "未找到相关文档。", "sources": []},
+                ensure_ascii=False,
+                indent=2,
+            )
+
+        context = "\n\n".join(
+            f"[{d['title']}] (ID: {d['id']}, 项目: {d['project']})\n{d['content']}"
+            for d in docs
+        )
+        prompt = f"根据以下资源文档回答用户问题。\n\n问题：{question}\n\n资源文档：\n{context}"
+
+        agent = get_agent()
+        answer = await agent.run(prompt)
+
+        return json.dumps(
+            {
+                "success": True,
+                "answer": answer.output,
+                "sources": [{"id": d["id"], "title": d["title"], "score": d["score"]} for d in docs],
+            },
+            ensure_ascii=False,
+            indent=2,
+            default=str,
+        )
+    except Exception as e:
+        logger.error(f"智能查询失败: {e}")
+        return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
+
+
+# ==================== 工具映射（供 API 使用）====================
+
+TOOL_MAP = {
+    "create_index": create_index,
+    "list_indexes": list_indexes,
+    "upload_documents": upload_documents,
+    "search_documents": search_documents,
+    "get_document": get_document,
+    "update_document": update_document,
+    "delete_documents": delete_documents,
+    "smart_query": smart_query,
+}
+
+TOOL_LIST = [
+    {
+        "name": "create_index",
+        "description": "创建支持混合搜索（关键词+向量）的 Azure AI Search 索引",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "index_name": {"type": "string", "description": "索引名称"},
+            },
+            "required": ["index_name"],
+        },
+    },
+    {
+        "name": "list_indexes",
+        "description": "列出所有可用的搜索索引",
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "upload_documents",
+        "description": "上传文档到搜索索引（自动生成向量，支持批量），每个文档需包含 id、title、content",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "documents": {"type": "string", "description": "JSON 格式文档数组"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+            },
+            "required": ["documents"],
+        },
+    },
+    {
+        "name": "search_documents",
+        "description": "混合搜索文档（关键词+向量语义），支持筛选、排序",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "搜索关键词或自然语言查询"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+                "filter_expr": {"type": "string", "description": "OData 筛选表达式（可选）"},
+                "top": {"type": "integer", "description": "返回数量", "default": 10},
+                "select": {"type": "string", "description": "返回字段，逗号分隔（可选）"},
+                "order_by": {"type": "string", "description": "排序字段（可选）"},
+                "search_mode": {
+                    "type": "string",
+                    "description": "搜索模式: hybrid（混合，默认）、keyword（纯关键词）、vector（纯向量）",
+                    "enum": ["hybrid", "keyword", "vector"],
+                    "default": "hybrid",
+                },
+            },
+            "required": ["query"],
+        },
+    },
+    {
+        "name": "get_document",
+        "description": "根据 ID 获取单个文档",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "document_id": {"type": "string", "description": "文档 ID"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+                "select": {"type": "string", "description": "返回字段，逗号分隔（可选）"},
+            },
+            "required": ["document_id"],
+        },
+    },
+    {
+        "name": "update_document",
+        "description": "更新文档部分字段（merge），更新 title/content 会自动重新生成向量",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "document_id": {"type": "string", "description": "文档 ID"},
+                "updates": {"type": "string", "description": "JSON 格式的更新字段"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+            },
+            "required": ["document_id", "updates"],
+        },
+    },
+    {
+        "name": "delete_documents",
+        "description": "根据 ID 删除一个或多个文档",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "document_ids": {"type": "string", "description": "文档 ID 列表 JSON 数组或单个 ID"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+            },
+            "required": ["document_ids"],
+        },
+    },
+    {
+        "name": "smart_query",
+        "description": "智能问答：混合搜索 + AI 总结，用自然语言提问并获取文档总结",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "question": {"type": "string", "description": "自然语言问题"},
+                "index_name": {"type": "string", "description": "索引名称（可选）"},
+                "project": {"type": "string", "description": "限定项目名称（可选）"},
+                "top": {"type": "integer", "description": "返回结果数量", "default": 5},
+            },
+            "required": ["question"],
+        },
+    },
+]
+
+
+if __name__ == "__main__":
+    server.run()