第 26 期 | 基于 MCP 的跨应用组合拳 (Orchestration)
🎯 学习目标
学完本期课程，你将能够：
- 理解并阐述基于大型语言模型（LLM）的跨应用编排（Orchestration）核心原理。
- 掌握如何为 Claude 配置和定义多平台 API 工具，使其能够与外部服务交互。
- 设计并实现一个多步骤、多工具的自动化工作流，以解决实际的跨平台协作场景。
- 提升利用 AI Agent 进行复杂任务自动化、提高开发效率和协作效率的能力。
📖 核心概念讲解
26.1 什么是基于 LLM 的跨应用编排 (Orchestration)?
在现代软件开发和产品管理流程中，我们经常需要在不同的工具和平台之间切换，例如从设计工具 Figma 下载素材，到项目管理工具 Linear 创建任务，再到代码托管平台 GitHub 提交代码。这些独立的步骤往往需要手动操作或复杂的脚本串联。
基于 LLM 的跨应用编排 (Orchestration)，其核心思想是利用大型语言模型（如 Claude）作为智能协调器，通过理解自然语言指令，动态地调用和组合多个外部工具（API），从而自动化执行跨越多个应用或服务的工作流。这里的 "MCP" 可以理解为 "Model Control Plane" 或 "Multi-platform Control Plane"，即 LLM 充当一个中心控制平面，统一管理和调度对不同外部平台的访问与操作。
与传统的集成方案（如 ETL 工具、中间件、Zapier/IFTTT 等）不同，LLM 驱动的编排具有以下显著优势：
- 意图理解与动态决策： LLM 能理解复杂、模糊的自然语言指令，并根据上下文动态地规划执行路径，而不是依赖预设的固定规则。
- 工具组合与链式调用： LLM 能够智能地将多个工具按逻辑顺序组合起来，前一个工具的输出可以作为后一个工具的输入，形成复杂的任务链。
- 自适应与错误处理： LLM 可以在一定程度上识别并尝试处理工具调用中出现的错误，甚至调整策略或请求用户澄清。
工作流示意图：
graph TD
A[用户自然语言指令] --> B(Claude Agent - Orchestrator)
B --> C{意图理解 & 任务分解}
C --> D[选择并调用工具: Figma API]
D -- 设计资产URL/数据 --> E[选择并调用工具: Linear API]
E -- 任务ID/URL --> F[选择并调用工具: GitHub API]
F -- 提交URL --> G[结果汇总 & 用户反馈]26.2 如何为 Claude 配置外部工具 (Tool Definitions)?
要让 Claude 能够与外部应用交互，我们需要为它提供“工具”的定义。这些工具本质上是封装了特定 API 调用的函数描述，Claude 会根据其理解的意图和这些工具的描述来决定何时调用哪个工具，以及传递什么参数。
Anthropic 的 Claude 提供了 tool_code 机制（或称为 Function Calling），允许开发者定义工具的名称、描述以及输入参数的 JSON Schema。当 Claude 认为某个工具能够帮助它完成用户请求时，它会生成一个符合该工具 input_schema 的 JSON 对象，然后由外部执行环境（通常是你的代码）来实际调用对应的 API。
工具定义的核心要素：
name: 工具的唯一标识符，Claude 会用它来引用工具。description: 详细描述工具的功能、用途和何时应该使用它。这对于 Claude 的决策至关重要。input_schema: 一个 JSON Schema 对象，定义了工具接受的参数及其类型、是否必填、描述等。
示例：一个简化的 Figma 下载工具的 input_schema
{
"name": "figma_download_asset",
"description": "下载 Figma 设计文件中的指定节点（如组件、画板、图层）为图片或SVG格式。需要 Figma 文件ID 和节点ID。",
"input_schema": {
"type": "object",
"properties": {
"file_id": {
"type": "string",
"description": "Figma 文件的唯一标识符。"
},
"node_id": {
"type": "string",
"description": "Figma 文件中要下载的节点的唯一标识符。"
},
"format": {
"type": "string",
"enum": ["png", "jpg", "svg", "pdf"],
"description": "下载的图片格式。"
},
"scale": {
"type": "number",
"description": "图片导出时的缩放比例，例如 1、2、3。",
"default": 1
}
},
"required": ["file_id", "node_id", "format"]
}
}
Claude 在接收到用户请求后，会根据其语言模型能力，结合这些工具定义进行推理，生成一个或多个 tool_use 块，其中包含它决定调用的工具名称和参数。你的应用程序负责捕获这些 tool_use 块，执行实际的 API 调用，并将结果（tool_result）返回给 Claude，以便它继续处理或给出最终响应。
26.3 编排策略：多步骤与错误处理
一个成功的跨应用组合拳，不仅仅是简单地按顺序调用工具，还需要考虑更复杂的编排策略：
顺序执行 (Sequential Execution): 这是最基本的策略，即工具按照逻辑顺序依次调用。例如，先下载 Figma 文件，再根据下载结果创建 Linear 任务，最后将文件推送到 GitHub。Claude 能够基于其对任务的理解自动推断这种顺序。
请求 -> Tool A (成功) -> Tool B (成功) -> Tool C (成功) -> 响应条件分支 (Conditional Logic): 在某些场景下，下一步的执行可能依赖于前一步的结果。例如，如果 Figma 文件下载失败，则不应尝试创建 Linear 任务和推送 GitHub，而是直接报告错误。Claude 可以通过分析
tool_result来做出这种条件判断。请求 -> Tool A (成功) -> IF (A的结果满足条件) THEN Tool B ELSE 报告错误循环与重试 (Loops & Retries): 对于一些可能暂时失败的操作，可以设置重试机制。虽然 Claude 本身没有内建的循环结构，但可以通过在
tool_result中返回特定的状态，引导 Claude 再次尝试调用工具，或者在外部执行环境中实现重试逻辑。状态管理 (State Management): 跨多个工具调用时，通常需要将一个工具的输出作为另一个工具的输入。例如，Figma 下载工具返回的资产 URL/内容，需要传递给 GitHub 推送工具。Claude 的上下文窗口使其能够“记住”之前工具调用的结果，并将其作为后续工具调用的参数。
错误处理与用户反馈 (Error Handling & User Feedback):
- 工具执行层面的错误： 当实际执行工具的代码遇到网络问题、API 认证失败、参数错误等情况时，应将详细的错误信息封装在
tool_result中返回给 Claude。 - Claude 的处理： Claude 接收到错误后，可以尝试：
- 解释错误原因。
- 建议用户如何解决。
- 如果可能，尝试使用备用工具或策略。
- 请求用户提供更多信息。
- 用户反馈： 在整个编排过程中，尤其是在长时间运行的任务中，及时向用户提供进度和状态更新至关重要。
- 工具执行层面的错误： 当实际执行工具的代码遇到网络问题、API 认证失败、参数错误等情况时，应将详细的错误信息封装在
通过精心设计工具定义和利用 Claude 的推理能力，我们可以构建出强大而灵活的跨应用自动化工作流。
💻 模拟 Claude TUI 交互
假设我们已经为 Claude 配置好了 figma_download_asset, linear_create_issue, github_push_file 这三个工具。
场景 1: 用户发起跨平台任务
> 用户：我需要你帮我处理一个新组件。请从 Figma ID 为 '123456789' 的文件中，下载节点 ID 'N:123:456' 的 PNG 格式图片。下载后，在 Linear 中创建一个名为“集成 Figma 组件：[组件名称]”的任务，并将其指派给我 (我的用户 ID 是 'usr_abcde')，项目 ID 是 'prj_xyz'。最后，将下载的 PNG 图片推送到 GitHub 仓库 'my-org/design-assets' 的 `assets/components/` 目录下，提交消息为“feat: Add [组件名称] from Figma”。请确保组件名称从 Figma 节点中获取。
╭─ Claude ──────────────────────────────╮
│ 好的，我明白了您的请求。这涉及到从 Figma 下载资产，在 Linear 创建任务，并将资产推送到 GitHub。│
│ 我将按照以下步骤进行： │
│ 1. 调用 figma_download_asset 工具，从 Figma 获取指定节点信息并下载 PNG 图片。 │
│ 2. 从 Figma 节点信息中提取组件名称。 │
│ 3. 调用 linear_create_issue 工具，在 Linear 中创建任务，并使用提取到的组件名称。 │
│ 4. 调用 github_push_file 工具，将下载的 PNG 图片推送到 GitHub。 │
│ │
│ 请稍等，我正在执行。 │
│ │
│ │
│ print(tool_figma_api.download_asset(file_id='123456789', node_id='N:123:456', format='png'))│
│ │
╰───────────────────────────────────────╯
场景 2: Claude 执行工具并进行链式调用
假设 figma_download_asset 工具成功执行，返回了资产的名称和下载内容的 Base64 编码。
╭─ Claude ──────────────────────────────╮
│ <tool_result> │
│ { │
│ "status": "success", │
│ "asset_name": "ButtonComponent.png", │
│ "content_base64": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=", │
│ "download_url": "https://example.com/figma_asset/ButtonComponent.png" │
│ } │
│ </tool_result> │
│ │
│ Figma 资产 'ButtonComponent.png' 已成功下载。现在我将在 Linear 中创建任务。 │
│ │
│ <tool_code> │
│ print(tool_linear_api.create_issue( │
│ title='集成 Figma 组件：ButtonComponent', │
│ description='从 Figma 文件 ID 123456789，节点 ID N:123:456 下载并集成 ButtonComponent.png。',│
│ project_id='prj_xyz', │
│ assignee_id='usr_abcde' │
│ )) │
│ </tool_code> │
╰───────────────────────────────────────╯
假设 linear_create_issue 工具成功执行。
╭─ Claude ──────────────────────────────╮
│ <tool_result> │
│ { │
│ "status": "success", │
│ "issue_id": "LIN-101", │
│ "issue_url": "https://linear.app/my-org/issue/LIN-101/集成-figma-组件-buttoncomponent" │
│ } │
│ </tool_result> │
│ │
│ Linear 任务 LIN-101 已成功创建。最后一步，我将把下载的图片推送到 GitHub。 │
│ │
│ <tool_code> │
│ print(tool_github_api.push_file( │
│ repo='my-org/design-assets', │
│ path='assets/components/ButtonComponent.png', │
│ content_base64='iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=', │
│ commit_message='feat: Add ButtonComponent from Figma', │
│ branch='main' │
│ )) │
│ </tool_code> │
╰───────────────────────────────────────╯
场景 3: 任务完成与结果反馈
假设 github_push_file 工具成功执行。
╭─ Claude ──────────────────────────────╮
│ <tool_result> │
│ { │
│ "status": "success", │
│ "commit_sha": "a1b2c3d4e5f6...", │
│ "file_url": "https://github.com/my-org/design-assets/blob/main/assets/components/ButtonComponent.png" │
│ } │
│ </tool_result> │
│ │
│ 太棒了！所有步骤都已成功完成。 │
│ │
│ - Figma 资产 'ButtonComponent.png' 已成功下载。 │
│ - Linear 任务 LIN-101 ('集成 Figma 组件：ButtonComponent') 已创建，详情请访问： │
│ https://linear.app/my-org/issue/LIN-101/集成-figma-组件-buttoncomponent │
│ - PNG 图片已成功推送到 GitHub 仓库 'my-org/design-assets' 的 `assets/components/ButtonComponent.png`。│
│ 文件链接：https://github.com/my-org/design-assets/blob/main/assets/components/ButtonComponent.png │
│ │
│ 还有其他我可以帮助您的吗？ │
╰───────────────────────────────────────╯
💻 代码演示
这里我们展示为 Claude 定义的三个核心工具的 JSON Schema，以及一个简化的 Python 伪代码，展示如何在你的后端服务中执行这些工具并与 Claude 交互。
1. 工具定义 (JSON Schema)
这些 JSON 对象将被作为 tools 参数传递给 Claude API。
figma_download_asset 工具定义:
// figma_tool_definition.json
{
"name": "figma_download_asset",
"description": "从 Figma 设计文件中下载指定节点的资产（如组件、画板、图层）为图片或SVG格式。此工具会返回资产的 Base64 编码内容和名称。",
"input_schema": {
"type": "object",
"properties": {
"file_id": {
"type": "string",
"description": "Figma 文件的唯一标识符。"
},
"node_id": {
"type": "string",
"description": "Figma 文件中要下载的节点的唯一标识符。"
},
"format": {
"type": "string",
"enum": ["png", "jpg", "svg", "pdf"],
"description": "下载的图片格式。"
},
"scale": {
"type": "number",
"description": "图片导出时的缩放比例，例如 1、2、3。",
"default": 1
}
},
"required": ["file_id", "node_id", "format"]
}
}
linear_create_issue 工具定义:
// linear_tool_definition.json
{
"name": "linear_create_issue",
"description": "在 Linear 中创建一个新的任务（Issue）。",
"input_schema": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "任务的标题。"
},
"description": {
"type": "string",
"description": "任务的详细描述。"
},
"project_id": {
"type": "string",
"description": "任务所属项目的唯一标识符。"
},
"assignee_id": {
"type": "string",
"description": "任务的指派人用户 ID。"
},
"priority": {
"type": "string",
"enum": ["No priority", "Low", "Medium", "High", "Urgent"],
"description": "任务的优先级。",
"default": "No priority"
}
},
"required": ["title", "description", "project_id", "assignee_id"]
}
}
github_push_file 工具定义:
// github_tool_definition.json
{
"name": "github_push_file",
"description": "将文件内容推送到 GitHub 仓库的指定路径。如果文件不存在则创建，如果存在则更新。",
"input_schema": {
"type": "object",
"properties": {
"repo": {
"type": "string",
"description": "GitHub 仓库的完整名称，例如 'owner/repo-name'。"
},
"path": {
"type": "string",
"description": "文件在仓库中的路径，例如 'assets/images/logo.png'。"
},
"content_base64": {
"type": "string",
"description": "文件的 Base64 编码内容。"
},
"commit_message": {
"type": "string",
"description": "Git 提交信息。"
},
"branch": {
"type": "string",
"description": "要推送到的分支名称，例如 'main' 或 'develop'。",
"default": "main"
}
},
"required": ["repo", "path", "content_base64", "commit_message"]
}
}
2. 模拟后端工具执行逻辑 (Python 伪代码)
这是一个简化的 Python 脚本片段，展示了你的应用程序如何接收 Claude 的 tool_use 请求，执行相应的工具函数，并将结果返回给 Claude。
import os
import requests
import base64
import json
from anthropic import Anthropic
# --- API Keys and Endpoints (实际使用时应从环境变量或安全配置中加载) ---
Figma_API_TOKEN = os.getenv("Figma_API_TOKEN")
LINEAR_API_KEY = os.getenv("LINEAR_API_KEY")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
Figma_API_BASE = "https://api.figma.com/v1"
LINEAR_API_ENDPOINT = "https://api.linear.app/graphql"
GITHUB_API_BASE = "https://api.github.com"
# --- 实际工具函数实现 ---
class FigmaTool:
def download_asset(self, file_id: str, node_id: str, format: str, scale: int = 1):
headers = {"X-Figma-Token": Figma_API_TOKEN}
# 1. 获取导出 URL
export_url_resp = requests.get(
f"{Figma_API_BASE}/images/{file_id}?ids={node_id}&format={format}&scale={scale}",
headers=headers
)
export_url_resp.raise_for_status()
export_urls = export_url_resp.json().get("images", {})
if not export_urls or node_id not in export_urls:
return {"status": "error", "message": "无法获取 Figma 导出 URL。"}
image_url = export_urls[node_id]
# 2. 下载图片内容
image_resp = requests.get(image_url)
image_resp.raise_for_status()
image_content = image_resp.content
content_base64 = base64.b64encode(image_content).decode('utf-8')
# 3. 获取节点名称 (可选，如果需要更准确的名称，可能需要调用 /files/:file_id/nodes API)
node_name_resp = requests.get(
f"{Figma_API_BASE}/files/{file_id}/nodes?ids={node_id}",
headers=headers
)
node_name_resp.raise_for_status()
node_data = node_name_resp.json().get("nodes", {}).get(node_id, {}).get("document", {})
asset_name = f"{node_data.get('name', 'untitled_asset')}.{format}"
return {
"status": "success",
"asset_name": asset_name,
"content_base64": content_base64,
"download_url": image_url
}
class LinearTool:
def create_issue(self, title: str, description: str, project_id: str, assignee_id: str, priority: str = "No priority"):
headers = {
"Authorization": f"Bearer {LINEAR_API_KEY}",
"Content-Type": "application/json"
}
graphql_query = """
mutation IssueCreate($title: String!, $description: String!, $projectId: String!, $assigneeId: String!, $priority: String!) {
issueCreate(
input: {
title: $title,
description: $description,
projectId: $projectId,
assigneeId: $assigneeId,
priority: $priority
}
) {
success
issue {
id
url
}
lastSyncId
}
}
"""
variables = {
"title": title,
"description": description,
"projectId": project_id,
"assigneeId": assignee_id,
"priority": priority
}
resp = requests.post(LINEAR_API_ENDPOINT, headers=headers, json={"query": graphql_query, "variables": variables})
resp.raise_for_status()
data = resp.json()
issue_data = data.get("data", {}).get("issueCreate", {})
if issue_data.get("success"):
issue = issue_data.get("issue", {})
return {
"status": "success",
"issue_id": issue.get("id"),
"issue_url": issue.get("url")
}
return {"status": "error", "message": data.get("errors", "未知错误")}
class GitHubTool:
def push_file(self, repo: str, path: str, content_base64: str, commit_message: str, branch: str = "main"):
headers = {
"Authorization": f"token {GITHUB_TOKEN}",
"Content-Type": "application/json"
}
url = f"{GITHUB_API_BASE}/repos/{repo}/contents/{path}"
# 尝试获取文件 SHA，判断是创建还是更新
sha = None
try:
get_resp = requests.get(url, headers=headers, params={"ref": branch})
if get_resp.status_code == 200:
sha = get_resp.json().get("sha")
except requests.exceptions.RequestException:
pass # 文件可能不存在
payload = {
"message": commit_message,
"content": content_base64,
"branch": branch
}
if sha:
payload["sha"] = sha # 更新文件需要提供 SHA
resp = requests.put(url, headers=headers, json=payload)
resp.raise_for_status()
data = resp.json()
return {
"status": "success",
"commit_sha": data.get("commit", {}).get("sha"),
"file_url": data.get("content", {}).get("html_url")
}
# --- Claude 交互逻辑 (核心编排部分) ---
def execute_tool_call(tool_name: str, tool_input: dict):
if tool_name == "figma_download_asset":
return FigmaTool().download_asset(**tool_input)
elif tool_name == "linear_create_issue":
return LinearTool().create_issue(**tool_input)
elif tool_name == "github_push_file":
return GitHubTool().push_file(**tool_input)
else:
return {"status": "error", "message": f"未知工具: {tool_name}"}