1. 项目概述这不是一个“搭个服务”的故事而是一次对AI工作流底层契约的重写“How I Built a Custom Databricks MCP Server to Power Agentic AI”——这个标题里藏着三个被日常讨论严重稀释的关键词Databricks、MCP和Agentic AI。很多人看到“Databricks”第一反应是“哦那个做数据湖和Spark的云平台”看到“Agentic AI”想到的是“能自己规划任务的LLM”至于“MCP”十有八九会误以为是“Master Control Program”或者某个冷门缩写。但在这个项目里MCP 指的是Model Command Protocol一个由社区自发演进、尚未被任何大厂官方收编的轻量级模型调用规范。它不依赖OpenAI的RESTful API那种固定schema也不走LangChain那种高度抽象的链式封装而是用极简的JSON-RPC 2.0语义定义了“发什么指令、带什么上下文、期待什么结构化响应”的最小公约数。我花掉整整六周时间不是为了在Databricks上跑通一个LLM推理API而是要把它变成一个可编程的AI代理调度中枢当业务系统发来一条“分析Q3销售下滑原因并生成PPT大纲”的请求时这个MCP Server必须能自动拆解为“查Snowflake销售表→调用Llama-3-70B做归因分析→调用Claude-3-Haiku生成PPT结构→用本地微调的LayoutLM校验格式合规性”这一串原子动作并确保每一步的输入输出都符合MCP定义的request_id、tool_call_id、execution_context等字段契约。为什么非得自己造因为Databricks原生的Model Serving只解决“单次推理”问题它的Endpoint本质是个无状态函数你无法让它记住前一步的中间结果更无法让多个模型像乐高一样按需拼装。而Agentic AI的核心痛点恰恰在于状态管理与工具路由——Agent需要维护对话历史、临时变量、执行栈还要根据当前语义动态选择调用哪个模型、哪个数据库、哪个外部API。市面上的方案要么太重如部署一套完整的LangGraph服务要么太散用Airflow调度一堆独立脚本。我最终选择在Databricks上构建MCP Server是因为它天然具备三重不可替代性第一统一权限层所有模型、数据、计算资源都通过Unity Catalog纳管避免在K8s集群里手动配置RBAC第二零拷贝数据流动Agent的中间结果比如从Delta表查出的原始销售数据可以直接作为下一个模型的输入无需序列化/反序列化到S3或Redis第三计算弹性当并发请求激增时Databricks的Auto-scaling Cluster能在90秒内从2个Worker扩到32个而自建K8s集群的HPA往往需要3分钟以上。这个项目不是炫技而是把Databricks从“数据处理引擎”升级为“AI代理操作系统”让每个业务分析师都能用SQL语法描述AI任务而不是写Python代码。2. 核心设计思路为什么放弃LangChain、为什么坚持MCP、为什么选Databricks UC作为唯一真相源2.1 放弃LangChain不是因为它不好而是因为它太“好”了LangChain的抽象层确实强大但它解决的是“如何让不同模型说话”的问题而Agentic AI真正卡脖子的是“如何让模型们协同做事”。我试过用LangChain的AgentExecutor封装Databricks Model Serving Endpoint很快撞上三堵墙第一状态丢失。AgentExecutor每次调用都会新建一个Runnable对象之前步骤生成的intermediate_steps比如已提取的客户ID列表在下一步骤中完全不可见除非你手动塞进chat_history但这会导致上下文长度爆炸第二工具注册僵化。LangChain要求所有Tool必须在初始化时静态注册而我们的业务场景需要动态加载工具——比如财务部门今天要接入新上线的ERP系统API明天要替换旧版信用评分模型这种高频变更在LangChain里意味着重启整个服务进程第三错误传播失真。当Llama-3在归因分析中返回格式错误的JSON时LangChain的parse_output方法会抛出OutputParserException但这个异常里不包含原始模型响应的request_id和model_name运维人员根本无法定位是哪个模型、哪次调用出了问题。我最终删掉了所有LangChain依赖不是因为它技术落后而是它的设计哲学与Agentic AI的实时性、动态性、可观测性需求存在根本冲突。2.2 MCP协议用12行JSON-RPC定义AI协作的宪法MCP协议的核心价值在于它用最简的语义约束解决了模型间协作的“信任问题”。它的完整规范只有一页Markdown但每一行都直击痛点。我们来看一个真实请求示例{ jsonrpc: 2.0, method: execute_tool, params: { tool_name: sales_analytics_query, arguments: { region: EMEA, quarter: Q3-2024, metric: revenue }, context: { request_id: req_abc123, parent_request_id: req_xyz789, execution_context: agent_sales_insight_v2 } }, id: 1 }这里的关键设计点在于context字段request_id是全局唯一追踪ID贯穿整个Agent生命周期parent_request_id实现调用链路的树状追溯execution_context则指定了本次调用所处的业务环境比如agent_sales_insight_v2会自动加载该版本对应的模型权重、数据权限、超时策略。对比OpenAI的APIMCP没有model、messages这些字段因为它默认所有工具调用都发生在同一个Agent会话内模型选择由Server端根据tool_name和execution_context动态路由。这种设计让客户端极度轻量——前端App只需发送标准JSON-RPC无需关心背后是调用Databricks上的Llama-3还是调用AWS Sagemaker上的Claude-3。我坚持MCP而非自定义协议是因为它已被Databricks Labs的几个开源项目采用这意味着未来可以无缝对接他们的databricks-mcp-server参考实现避免陷入“重复造轮子”的陷阱。2.3 Unity Catalog不是数据库目录而是AI时代的“服务注册中心”很多人把Unity Catalog当成一个高级版的Hive Metastore但在这个项目里它承担着远超元数据管理的角色。我把所有AI能力都注册为UC中的“模型”Model或“函数”Functionmodels.sales_insight.llama3_70b_v2指向Databricks Model Serving的Endpoint URL附带inference_params如temperature0.3,max_tokens2048functions.data_tools.sales_analytics_query一个SQL函数封装了查询Salesforce和NetSuite数据的复杂JOIN逻辑functions.validation.layoutlm_ppt_checker一个Python UDF调用微调后的LayoutLM模型校验PPT大纲格式关键创新在于我为每个UC对象添加了自定义Tagmcp_tool: true、execution_context: agent_sales_insight_v2、timeout_seconds: 120。当MCP Server收到execute_tool请求时它不解析硬编码的路由表而是直接查询UCSELECT catalog, schema, name, properties[inference_params] as inference_params, properties[timeout_seconds] as timeout_seconds FROM system.information_schema.models WHERE properties[mcp_tool] true AND properties[execution_context] agent_sales_insight_v2 AND name llama3_70b_v2这种设计带来三大优势第一权限即代码数据科学家给models.sales_insight.llama3_70b_v2赋予权限就等于赋予了Agent调用该模型的能力第二配置即数据修改超时时间只需更新UC属性无需重启服务第三发现即集成新开发的工具只要打上mcp_tool: true标签就会被Server自动识别。Unity Catalog在这里不再是数据目录而是整个AI代理生态的“服务注册中心”和“策略控制平面”。3. 核心模块实现从MCP Server骨架到生产级容错的完整落地3.1 MCP Server基础骨架用FastAPIDatabricks SDK构建最小可行服务MCP Server的入口是一个标准的FastAPI应用但它的核心逻辑与传统Web服务截然不同。我刻意避开了ASGI中间件的复杂封装选择用最朴素的方式实现JSON-RPC 2.0的语义解析。以下是main.py的核心片段from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel import json import logging from databricks.sdk import WorkspaceClient from databricks.sdk.service.catalog import FunctionInfo app FastAPI(titleDatabricks MCP Server) # 初始化Databricks客户端使用Service Principal认证 dbx_client WorkspaceClient( hosthttps://workspace-id.cloud.databricks.com, tokenservice-principal-token ) class MCPRequest(BaseModel): jsonrpc: str method: str params: dict id: int app.post(/mcp) async def handle_mcp_request(request: Request): try: # 1. 严格验证JSON-RPC 2.0格式 raw_body await request.body() payload json.loads(raw_body) if payload.get(jsonrpc) ! 2.0: raise HTTPException(400, Invalid JSON-RPC version) # 2. 提取核心上下文用于日志追踪和策略匹配 context payload[params].get(context, {}) request_id context.get(request_id, unknown) execution_context context.get(execution_context, default) # 3. 根据method分发到具体处理器 if payload[method] execute_tool: return await execute_tool_handler(payload, request_id, execution_context) elif payload[method] list_tools: return await list_tools_handler(execution_context) else: raise HTTPException(400, fUnsupported method: {payload[method]}) except json.JSONDecodeError as e: logging.error(fJSON decode error for request {request_id}: {e}) raise HTTPException(400, Invalid JSON format) except Exception as e: logging.error(fUnexpected error for request {request_id}: {e}) raise HTTPException(500, Internal server error)这个骨架看似简单但每个细节都经过深思熟虑。首先不使用request.json()而是显式调用await request.body()再json.loads()因为这样可以捕获原始字节流便于后续做审计日志和恶意Payload检测其次上下文提取前置在进入具体处理器前就拿到request_id和execution_context确保所有日志、监控、策略检查都有统一的追踪标识最后错误分类处理JSON解析错误返回400业务逻辑错误返回500这为前端提供了明确的重试策略依据。我测试过这个骨架在单节点上能稳定处理1200 RPS瓶颈不在Python本身而在Databricks SDK的HTTP连接池配置——将max_connections从默认的10提升到100后吞吐量直接翻倍。3.2 工具动态发现与路由用Unity Catalog元数据驱动运行时决策execute_tool_handler是整个Server的“大脑”它的核心任务是根据tool_name和execution_context从Unity Catalog中找到对应工具的完整定义并决定调用方式。这个过程分为四步第一步UC元数据查询def get_tool_definition(tool_name: str, execution_context: str) - dict: 从Unity Catalog查询工具定义支持模型和函数两种类型 返回标准化的tool_spec字典包含type、endpoint、params等字段 # 查询Catalog中名为tool_name的模型 try: model dbx_client.models.get(f{tool_name}) if model.properties.get(mcp_tool) true: return { type: model, endpoint: model.serving_endpoint, inference_params: json.loads(model.properties.get(inference_params, {})), timeout: int(model.properties.get(timeout_seconds, 60)) } except Exception as e: pass # 模型不存在继续查函数 # 查询Catalog中名为tool_name的函数 try: func_info dbx_client.functions.get(f{tool_name}) if func_info.properties.get(mcp_tool) true: return { type: function, catalog: func_info.catalog_name, schema: func_info.schema_name, name: func_info.name, timeout: int(func_info.properties.get(timeout_seconds, 30)) } except Exception as e: pass raise HTTPException(404, fTool {tool_name} not found in UC for context {execution_context})第二步动态路由决策async def route_to_tool(tool_spec: dict, arguments: dict, request_id: str) - dict: 根据tool_spec.type决定调用路径 - typemodel: 调用Databricks Model Serving REST API - typefunction: 执行SQL函数调用 if tool_spec[type] model: return await call_model_endpoint(tool_spec, arguments, request_id) elif tool_spec[type] function: return await call_sql_function(tool_spec, arguments, request_id) else: raise HTTPException(500, fUnknown tool type: {tool_spec[type]})第三步模型调用的健壮封装import aiohttp import asyncio async def call_model_endpoint(tool_spec: dict, arguments: dict, request_id: str) - dict: 封装对Databricks Model Serving Endpoint的异步调用 包含重试、超时、错误映射等生产级特性 endpoint_url f{tool_spec[endpoint]}/invocations headers { Authorization: fBearer {DBX_TOKEN}, Content-Type: application/json } # 构建请求体兼容Databricks的served model schema payload { dataframe_split: { columns: [input], data: [[json.dumps(arguments)]] } } timeout aiohttp.ClientTimeout(totaltool_spec[timeout]) async with aiohttp.ClientSession(timeouttimeout) as session: for attempt in range(3): # 最多重试3次 try: async with session.post(endpoint_url, headersheaders, jsonpayload) as resp: if resp.status 200: result await resp.json() # 解析Databricks返回的dataframe_split格式 output result[predictions][0] return { status: success, output: output, request_id: request_id, tool_name: tool_spec[endpoint].split(/)[-1] } elif resp.status 429: # 遇到限流指数退避 await asyncio.sleep(2 ** attempt random.uniform(0, 1)) continue else: raise HTTPException(resp.status, fModel call failed: {resp.reason}) except asyncio.TimeoutError: if attempt 2: raise HTTPException(504, Model call timeout after retries) await asyncio.sleep(2 ** attempt) raise HTTPException(500, Unexpected error in model call)第四步SQL函数的安全调用from pyspark.sql import SparkSession def call_sql_function(tool_spec: dict, arguments: dict, request_id: str) - dict: 在Databricks Runtime中安全执行SQL函数 使用参数化查询防止SQL注入 spark SparkSession.builder.getOrCreate() # 构建参数化SQL所有arguments都转为字符串并用单引号包裹 # 这里做了简化实际生产中会用更严格的类型转换 arg_str , .join([f{v} for v in arguments.values()]) sql fSELECT * FROM {tool_spec[catalog]}.{tool_spec[schema]}.{tool_spec[name]}({arg_str}) try: result_df spark.sql(sql) # 转换为字典列表适配MCP响应格式 result_list [row.asDict() for row in result_df.collect()] return { status: success, output: result_list, request_id: request_id, tool_name: f{tool_spec[catalog]}.{tool_spec[schema]}.{tool_spec[name]} } except Exception as e: logging.error(fSQL function call failed for {request_id}: {e}) raise HTTPException(500, fSQL function execution error: {str(e)})这套机制让工具管理彻底脱离代码运维人员只需在Databricks UI里更新UC对象的properties就能完成模型切换、超时调整、权限变更。我曾亲眼看到数据工程师在下午3点更新了models.sales_insight.llama3_70b_v2的inference_params3点02分所有Agent请求就自动开始使用新的temperature参数全程零发布、零中断。3.3 状态管理与会话持久化用Delta Table实现轻量级Agent状态机Agentic AI最棘手的问题是状态管理。如果每个HTTP请求都是无状态的那么Agent就无法记住“我已经查了Q3销售数据现在要对这些数据做分析”。我的解决方案是用Delta Table作为分布式会话存储每个request_id对应一行记录字段包括session_stateJSON字符串、last_updated、ttl_seconds。-- 创建Agent Session状态表 CREATE TABLE IF NOT EXISTS catalog.schema.agent_sessions ( request_id STRING COMMENT 全局唯一请求ID, session_state STRING COMMENT JSON格式的会话状态包含变量、执行栈等, last_updated TIMESTAMP COMMENT 最后更新时间, ttl_seconds INT COMMENT TTL单位秒0表示永不过期, created_at TIMESTAMP COMMENT 创建时间 ) USING DELTA PARTITIONED BY (date(created_at)) TBLPROPERTIES (delta.autoOptimize.optimizeWrite true);在execute_tool_handler中每次调用前先读取状态调用后更新状态def load_session_state(request_id: str) - dict: 从Delta表加载会话状态 try: df spark.sql(fSELECT session_state FROM catalog.schema.agent_sessions WHERE request_id {request_id}) if df.count() 0: return json.loads(df.first()[session_state]) except Exception as e: logging.warning(fFailed to load session state for {request_id}: {e}) return {} def save_session_state(request_id: str, state: dict, ttl_seconds: int 300): 保存会话状态到Delta表自动处理TTL from datetime import datetime, timedelta now datetime.now() expires_at now timedelta(secondsttl_seconds) if ttl_seconds 0 else None # 使用MERGE语句实现upsert避免并发写入冲突 merge_sql f MERGE INTO catalog.schema.agent_sessions t USING (SELECT {request_id} as request_id, {json.dumps(state)} as session_state, {now} as last_updated, {ttl_seconds} as ttl_seconds, {now} as created_at) s ON t.request_id s.request_id WHEN MATCHED THEN UPDATE SET t.session_state s.session_state, t.last_updated s.last_updated, t.ttl_seconds s.ttl_seconds WHEN NOT MATCHED THEN INSERT * spark.sql(merge_sql) # 在execute_tool_handler中使用 session_state load_session_state(request_id) # ... 执行工具调用可能修改session_state ... save_session_state(request_id, session_state, ttl_seconds300)这个设计巧妙利用了Delta Lake的ACID事务和MERGE语义解决了分布式环境下的状态竞争问题。更重要的是它让状态管理变得“可观察”——数据分析师可以直接用SQL查询agent_sessions表看到某个request_id的完整执行轨迹而不需要翻阅分散的日志文件。我设置默认TTL为300秒5分钟因为Agentic任务通常在这个时间内完成超时未完成的任务会被后台Job自动清理。3.4 生产级可观测性从日志、指标到分布式追踪的全链路覆盖一个无法观测的AI服务就是一颗定时炸弹。我为MCP Server构建了三层可观测性体系第一层结构化日志Logging所有关键操作都输出JSON格式日志包含request_id、tool_name、duration_ms、status等字段{ timestamp: 2024-06-15T14:23:45.123Z, level: INFO, request_id: req_abc123, event: tool_execution_start, tool_name: sales_analytics_query, arguments: {region: EMEA, quarter: Q3-2024}, execution_context: agent_sales_insight_v2 }这些日志被自动采集到Databricks的system.access表中可通过SQL实时分析“过去一小时哪个tool_name的平均延迟最高”、“哪些request_id在执行了3次后仍失败”。第二层Prometheus指标Metrics暴露标准Prometheus端点监控四大黄金信号mcp_server_requests_total{method, status_code}请求总量mcp_server_request_duration_seconds_bucket{le, tool_name}请求延迟分布mcp_server_tool_calls_total{tool_name, status}各工具调用次数mcp_server_session_cache_hit_ratio会话状态缓存命中率这些指标被Grafana可视化我设置了关键告警当mcp_server_request_duration_seconds_bucket{le2.0, tool_namellama3_70b_v2}的95分位超过2秒时立即触发Slack通知。第三层OpenTelemetry分布式追踪Tracing集成OpenTelemetry Python SDK为每个request_id生成完整的TraceSpan 1: handle_mcp_request (root) ├─ Span 2: get_tool_definition │ └─ Span 3: UC API call ├─ Span 4: route_to_tool │ ├─ Span 5: call_model_endpoint │ │ └─ Span 6: Databricks Model Serving call │ └─ Span 7: call_sql_function └─ Span 8: save_session_state └─ Span 9: Delta MERGE operation所有Span都携带request_id作为trace_id可在Jaeger UI中一键下钻看到某次失败请求的完整调用链精准定位是UC查询慢、还是模型Endpoint超时、还是SQL函数执行报错。这套可观测性体系让我在上线首周就发现了两个关键问题一是sales_analytics_query函数在处理EMEA区域数据时因JOIN条件未加索引导致平均延迟飙升至8秒二是llama3_70b_v2模型在temperature0.8时JSON解析失败率高达12%。没有这套体系这些问题可能要等用户投诉后才能发现。4. 实操部署与运维从本地开发到生产环境的平滑迁移4.1 本地开发环境用Docker Compose模拟Databricks全栈在把代码推送到生产集群前我必须确保它能在本地复现所有行为。为此我构建了一个精简的Docker Compose环境包含三个服务# docker-compose.yml version: 3.8 services: mcp-server: build: . ports: - 8000:8000 environment: - DBX_HOSThttps://fake-workspace.cloud.databricks.com - DBX_TOKENdummy-token depends_on: - uc-mock - model-serving-mock uc-mock: image: python:3.11-slim command: python -m http.server 8080 volumes: - ./mock-uc:/usr/src/app ports: - 8080:8080 model-serving-mock: image: python:3.11-slim command: python -m http.server 8081 volumes: - ./mock-model-serving:/usr/src/app ports: - 8081:8081uc-mock服务提供一个静态的HTTP接口模拟Unity Catalog的GET /api/2.1/unity-catalog/models/{full_name}返回预定义的JSON响应model-serving-mock则模拟Databricks Model Serving的POST /invocations根据请求参数返回不同的模拟结果成功、超时、格式错误。这样我在MacBook上就能完整测试MCP Server的所有分支逻辑无需连接真实的Databricks环境。开发效率提升了3倍因为我不再需要等待CI/CD流水线跑完才能验证一个简单的超时逻辑。4.2 CI/CD流水线GitOps驱动的自动化发布我采用GitOps模式管理生产部署所有配置和代码都存放在GitHub仓库中CI/CD由Databricks Workflows驱动# .github/workflows/deploy.yml name: Deploy MCP Server on: push: branches: [main] paths: - src/** - infra/** jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - name: Setup Python uses: actions/setup-pythonv4 with: python-version: 3.11 - name: Install dependencies run: pip install -r requirements.txt - name: Run unit tests run: pytest tests/ --covsrc/ - name: Build and push Docker image uses: docker/build-push-actionv4 with: push: true tags: ${{ secrets.DOCKER_REGISTRY }}/mcp-server:${{ github.sha }} - name: Trigger Databricks Workflow uses: databricks/github-actionv1 with: host: ${{ secrets.DATABRICKS_HOST }} token: ${{ secrets.DATABRICKS_TOKEN }} workflow-name: Deploy MCP Server parameters: | { image_tag: ${{ github.sha }}, cluster_id: ${{ secrets.PROD_CLUSTER_ID }} }Databricks Workflows中的Deploy MCP Server任务会执行一个Notebook该Notebook负责从ECR拉取最新Docker镜像更新Databricks上的Cluster Policy确保Worker节点满足GPU需求ml.g4dn.xlarge重启运行MCP Server的Interactive Cluster并挂载新的Docker镜像运行健康检查脚本向/mcp端点发送list_tools请求验证服务可用性发送Slack通知告知部署结果。整个流程从代码提交到服务上线平均耗时4分32秒。最关键的是回滚只需一次Git revert——将main分支回退到上一个commitCI/CD会自动触发旧版本部署无需人工干预。4.3 生产环境调优从内存泄漏到GPU利用率的深度优化上线初期我遇到了两个典型生产问题它们的解决过程极具代表性问题一内存泄漏导致OOM Killer频繁杀进程现象Server运行24小时后RSS内存持续增长最终被Linux OOM Killer终止。通过ps aux --sort-%mem发现uvicorn进程内存占用从200MB涨到4GB。 根因分析Python的aiohttpClientSession在异常退出时未正确关闭TCP连接导致大量TIME_WAIT状态的socket堆积进而耗尽内存。 解决方案强制在每次请求后显式关闭session并增加连接池大小# 修复前 async with aiohttp.ClientSession() as session: # ... 请求逻辑 ... # 修复后 session aiohttp.ClientSession( connectoraiohttp.TCPConnector( limit100, # 连接池大小 limit_per_host30, keepalive_timeout30 ) ) try: async with session.post(...) as resp: # ... 处理响应 ... finally: await session.close() # 显式关闭效果内存占用稳定在300MB左右72小时无波动。问题二GPU利用率长期低于20%现象nvidia-smi显示V100 GPU的util%平均只有15%但业务请求延迟却很高。 根因分析Llama-3-70B模型的batch size设置为1每次只处理单个请求GPU大部分时间在等待I/O。 解决方案在Databricks Model Serving中启用动态批处理Dynamic Batching并将max_batch_size设为8-- 在Model Serving配置中 SET serving_config { served_models: [{ name: llama3_70b_v2, model_name: models.sales_insight.llama3_70b_v2, workload_type: GPU_SMALL, scale_to_zero_enabled: true, config: { max_batch_size: 8, max_latency_ms: 2000 } }]};效果GPU利用率提升至65%-75%P95延迟从3.2秒降至1.4秒成本降低40%相同吞吐量下所需GPU实例数减少。这两个问题的解决印证了一个真理AI服务的性能瓶颈往往不在模型本身而在基础设施的精细调优。一个合格的MCP Server工程师必须同时是Python性能专家、Databricks配置专家和GPU硬件专家。5. 常见问题与实战排障那些文档里不会写的血泪教训5.1 “工具调用总是返回404但UC里明明存在”——权限继承的隐秘陷阱现象get_tool_definition函数总是在UC查询中返回空但我在Databricks UI里能清晰看到models.sales_insight.llama3_70b_v2这个模型。排查过程首先确认Service Principal的Token有效用curl -H Authorization: Bearer $TOKEN https://workspace.cloud.databricks.com/api/2.0/unity-catalog/models能正常列出所有模型然后检查dbx_client.models.get()的调用发现它内部调用的是GET /api/2.1/unity-catalog/models/{full_name}而full_name的格式必须是catalog.schema.name最终发现问题我在UC中创建模型时用的是CREATE MODEL sales_insight.llama3_70b_v2但Databricks默认将其full_name设为main.sales_insight.llama3_70b_v2main是默认catalog而我的代码里传入的是sales_insight.llama3_70b_v2。根本原因Unity Catalog的权限模型是分层的USE CATALOG权限必须显式授予Service Principal否则它只能看到maincatalog下的对象。而dbx_client.models.get()方法要求传入完整的catalog.schema.name不能省略catalog。解决方案方案A推荐在代码中始终使用catalog.schema.name格式例如models.main.sales_insight.llama3_70b_v2方案B为Service Principal授予USE CATALOG main权限并在UC中将模型创建在maincatalog下方案C在get_tool_definition中增加catalog自动发现逻辑遍历所有有权限的catalog。我选择了方案A因为它最明确、最不易出错。这个教训告诉我在Databricks世界里“默认”是最危险的假设。每一个API调用都必须严格对照官方文档的URL路径和参数要求不能凭经验猜测。5.2 “Agent执行到一半就卡住日志里没有任何错误”——Delta Table的隐式锁竞争现象某个复杂的Agentic任务涉及5个工具调用在第3步后停止响应agent_sessions表中该request_id的last_updated时间戳停留在那里后续请求全部超时。排查过程查看mcp-server日志发现call_sql_function调用后没有save_session_state的日志说明卡在了save_session_state的MERGE语句登录Databricks SQL Editor执行DESCRIBE HISTORY catalog.schema.agent_sessions发现最近一次MERGE操作耗时120秒且状态为RUNNING进一步执行SELECT * FROM system.access.logs WHERE event query AND duration_ms 100000定位到长时间运行的查询终极诊断SHOW TRANSACTIONS发现有一个MERGE事务处于ACTIVE状态持有catalog.schema.agent_sessions表的EXCLUSIVE锁。根因分析Delta Lake的MERGE操作在高并发下会产生写锁。当多个Agent请求共享同一个request_id比如前端重试时未生成新ID它们会同时尝试MERGE同一行导致锁等待。而我的save_session_state函数没有设置timeout导致一个请求卡死阻塞了整个Worker线程。解决方案在MERGE语句中加入LIMIT 1避免全表扫描为save_session_state添加超时和重