Choose your path
比較 MCP 與函式呼叫，並使用 CometAPI 建立一個完整的工具迴圈。依照 Python 程式碼，分別直接執行與透過 MCP 執行相同的訂單查詢。
from openai import OpenAI
client = OpenAI(
api_key="YOUR_COMETAPI_KEY",
base_url="https://api.cometapi.com/v1",
)
response = client.chat.completions.create(
model="gpt-5-mini",
messages=[{"role": "user", "content": "Build this workflow."}],
)
print(response.choices[0].message.content)函式呼叫 為模型提供請求工具的方式。MCP 為應用程式提供一種標準方式，用於在伺服器上探索與呼叫工具。若你的應用程式僅有少量自有工具，請使用直接函式整合；當你需要消費既有的相容服務，或透過共享介面暴露工具時，再加入 MCP。下文將透過 CometAPI，以單一完整代理迴圈示範同一個訂單查詢的兩種接法。
客服代理可以解釋出貨延遲，但要告知客戶包裹位置，就需要存取訂單系統。函式呼叫為模型提供請求該查詢的方式；你的應用程式負責執行查詢並回傳結果。處理器可使用本機函式庫、資料庫或遠端 API。
若要透過 MCP 來完成該請求，應用程式會使用用戶端以 tools/list 探索伺服器工具，並以 tools/call 呼叫它們。應用程式仍然決定要向模型暴露哪些已發現的工具。
圖 1. MCP 為 AI 應用程式與外部系統之間提供共同介面。來源：MCP 官方介紹。© 2024–2025 Anthropic, PBC 及其貢獻者。根據 MIT 授權條款未更動地重製；完整圖片來源與許可通知。
| 比較 | 函式呼叫 | MCP |
|---|---|---|
| 連線方式 | 模型對應用程式程式碼所發出的動作請求 | 應用程式的 MCP 用戶端連向工具伺服器 |
| 主要角色 | 允許模型以結構化參數請求具名操作 | 標準化相容應用程式發現並呼叫伺服器工具的方式 |
| 工具定義 | 以模型 API 的格式提供定義 | 由伺服器提供的定義，包含 inputSchema |
| 執行位置 | 應用程式分派到的處理器，可能呼叫本機或遠端服務 | MCP 伺服器上的工具實作，可能呼叫後端服務 |
| 是否需要獨立伺服器？ | 函式呼叫介面本身不需要伺服器 | 需要 MCP 伺服器，但用戶端可作為本機 stdio 子行程啟動；亦可選擇部署為遠端服務 |
| 典型用法 | 少量由單一應用程式擁有的工具，直接分派 | 既有的相容服務，或可跨多個應用程式重用的工具介面 |
| 能否一起運作？ | 應用程式可將模型的函式請求經由 MCP 路由 | MCP 用戶端以 tools/call 完成該請求並將結果回傳給應用程式 |
最後一行是兩者共建的關鍵。模型可以請求 get_order_status，而應用程式可選擇呼叫自身處理器，或改經由 MCP 伺服器。面向模型的函式定義可保持不變。
當你擁有少量只被單一應用程式使用的工具時，我們建議從直接整合開始。部署更少，且從模型請求到商務邏輯的路徑清晰易懂。
當已有相容的伺服器提供該能力，或多個 AI 應用程式需要共同的工具介面時，MCP 就很有用。相同的訂單工具可以同時服務客服助理與內部營運應用，各自控制其使用者存取方式。
一個分派器可以同時支援內部函式與 MCP 服務，將每個操作路由至對應的實作。函式庫與 REST API 也能共用；當 MCP 的標準探索與呼叫介面對工具消費方有助益時，就選擇 MCP。
我們將使用一個唯讀操作 get_order_status 與一筆虛構訂單 ORD-104。範例會先直接呼叫該操作，再透過本機 MCP 伺服器執行。兩種模式皆使用 CometAPI 的 Responses API 進行推論。
圖 2. 一個代理迴圈同時支援直接與 MCP 的執行路徑。原創示意，基於 MCP 架構與 CometAPI 函式呼叫流程。
使用 Python 3.11 或以上版本，以及 CometAPI 的 API 金鑰。本操作步驟以 Python 3.12.14 與 mcp==2.3.0 測試通過。依照 CometAPI 快速開始 建立金鑰。在 macOS 或 Linux shell 中建立專案並啟用虛擬環境：
mkdir mcp-agent-demo
cd mcp-agent-demo
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "mcp==2.3.0"
export COMETAPI_KEY="YOUR_COMETAPI_KEY"
以你的金鑰取代佔位字串，並避免將其置於公開原始碼中。此教學鎖定所用的 MCP SDK；官方 Python 伺服器指南 針對 2.x API。下列兩個代理命令會向 CometAPI 發送請求並消耗 API 點數。
將以下內容儲存為 orders.py。此處使用固定資料以保持教學自洽；之後可將查詢替換為你的實際服務。
"""Read-only, invented order data; never connects to a real order system."""
import re
def get_order_status(order_id: str) -> dict[str, str]:
"""Look up an invented order; ORD-104 is the only available record."""
if not isinstance(order_id, str) or not re.fullmatch(r"ORD-[0-9]{3}", order_id):
raise ValueError("order_id must have the form ORD-104")
if order_id != "ORD-104":
raise ValueError("Order not found in this demo")
return {"order_id": order_id, "status": "shipped", "source": "fictional_demo"}
呼叫 get_order_status("ORD-104") 會回傳此用例固定結果：
{"order_id": "ORD-104", "status": "shipped", "source": "fictional_demo"}
將以下完整腳本儲存為 agent.py。它會註冊函式、將問題送至 CometAPI、執行請求的呼叫，並以原始 call_id 回傳每一次的結果。
面向模型的定義使用 CometAPI Responses API 的 parameters 欄位。MCP 會提供 inputSchema；此範例在暴露已審查函式前，會將發現到的契約與僅含單一字串 order_id 的綱要比對。
"""A deliberately narrow direct/MCP bridge, not a general MCP adapter."""
import argparse
import asyncio
import json
import os
from pathlib import Path
import sys
from urllib.request import Request, urlopen
from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client
from orders import get_order_status
NAME = "get_order_status"
SCHEMA = {"type": "object", "properties": {"order_id": {"type": "string"}},
"required": ["order_id"], "additionalProperties": False}
TOOL = {"type": "function", "name": NAME, "description": "Look up fictional order ORD-104.",
"parameters": SCHEMA, "strict": True}
def validate_arguments(arguments):
# Exact validation for SCHEMA above; deliberately not a JSON Schema engine.
if (not isinstance(arguments, dict) or set(arguments) != {"order_id"}
or not isinstance(arguments["order_id"], str)):
raise ValueError("Expected exactly one string argument: order_id")
def validate_catalog(tools):
if len(tools) != 1 or tools[0].name != NAME:
raise RuntimeError("Unexpected MCP tool catalog")
schema = tools[0].input_schema
if (schema.get("type") != "object" or schema.get("required") != ["order_id"]
or set(schema.get("properties", {})) != {"order_id"}
or schema["properties"]["order_id"].get("type") != "string"):
raise RuntimeError("MCP schema does not match this reviewed demo tool")
async def dispatch(name, arguments, client=None):
if name != NAME:
raise ValueError("Tool is not allowlisted")
try:
validate_arguments(arguments)
if client is None:
return {"is_error": False, "data": get_order_status(**arguments)}
except ValueError as error:
return {"is_error": True, "error": str(error)}
result = await client.call_tool(name, arguments)
if getattr(result, "result_type", "complete") != "complete":
raise RuntimeError("Input-required MCP results are outside this demo")
if any(block.type != "text" for block in result.content):
raise RuntimeError("Non-text MCP content is outside this demo")
return {"is_error": bool(result.is_error),
"content": [block.text for block in result.content],
"structured_content": result.structured_content}
async def run_agent(send, execute, question, max_rounds=4):
history = [{"role": "user", "content": question}]
seen_call_ids = set()
for _ in range(max_rounds):
response = send({"model": "gpt-6-astra", "input": list(history),
"tools": [TOOL], "reasoning": {"effort": "low"}})
if response.get("status") != "completed" or response.get("error"):
raise RuntimeError("Model response failed or did not complete")
output = response["output"]
# CometAPI's documented manual replay: keep ALL items, omit top-level ids.
history.extend({key: value for key, value in item.items() if key != "id"}
for item in output)
blocks = [block for item in output if item["type"] == "message"
for block in item.get("content", [])]
if any(block["type"] == "refusal" for block in blocks):
raise RuntimeError("Model refused the request")
calls = [item for item in output if item["type"] == "function_call"]
if not calls:
text = "\n".join(block["text"] for block in blocks
if block["type"] == "output_text")
if not text:
raise RuntimeError("No final text returned")
return text
for call in calls:
call_id = call["call_id"]
if not isinstance(call_id, str) or not call_id or call_id in seen_call_ids:
raise ValueError("Missing or duplicate function call_id")
seen_call_ids.add(call_id)
try:
arguments = json.loads(call["arguments"])
except (ValueError, TypeError):
result = {"is_error": True, "error": "Arguments must be JSON text"}
else:
result = await execute(call["name"], arguments)
print(json.dumps({"tool": call["name"], "arguments": call["arguments"],
"result": result}), file=sys.stderr)
history.append({"type": "function_call_output", "call_id": call_id,
"output": json.dumps(result)})
raise RuntimeError("Tool loop reached its round limit")
def send_live(payload):
key = os.environ["COMETAPI_KEY"] # Only live mode reads the caller's key.
request = Request("https://api.cometapi.com/v1/responses",
data=json.dumps(payload).encode(),
headers={"Authorization": f"Bearer {key}",
"Content-Type": "application/json"})
with urlopen(request, timeout=45) as response:
return json.load(response)
def server_parameters():
return StdioServerParameters(command=sys.executable,
args=[str(Path(__file__).with_name("server.py"))])
async def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--mode", choices=["direct", "mcp"], required=True)
parser.add_argument("--question", default="Use the tool to check order ORD-104.")
args = parser.parse_args()
if args.mode == "direct":
answer = await run_agent(send_live, dispatch, args.question)
else:
async with Client(stdio_client(server_parameters())) as client:
discovered = await client.list_tools()
validate_catalog(discovered.tools)
async def execute(name, arguments):
return await dispatch(name, arguments, client)
answer = await run_agent(send_live, execute, args.question)
print(answer)
if __name__ == "__main__":
asyncio.run(main())
此迴圈會保留原始問題與每一個回傳的回應項目。依照 CometAPI Responses 的對話格式，在重播前移除輸出項目的頂層 id 欄位，並為每次呼叫附加一筆 function_call_output。保留其餘回傳項目也能保留下一回合所需的推理項目。
執行：
python agent.py --mode direct
應用程式會向模型描述 get_order_status。當模型請求該工具時，dispatch() 會檢查名稱與引數，然後呼叫 orders.py 中的函式。結果被寫回對話中，模型據此生成答案。
使用終端列印的工具追蹤確認查詢使用了 ORD-104，並回傳 status: shipped。最終答案應反映該結果；措辭可能不同。至此你已完成函式呼叫迴圈：從使用者的問題，到應用程式執行操作，再到模型回覆。
下一步將保持模型與對話迴圈不變，改把執行放到 MCP 伺服器後方，讓應用程式透過用戶端探索並呼叫該工具。
將以下內容儲存為 server.py，與前兩個檔案放在一起：
"""One read-only MCP tool, served over stdio. No stdout logging."""
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from orders import get_order_status as lookup_order
server = MCPServer("fictional-orders")
@server.tool(structured_output=True)
def get_order_status(order_id: str) -> dict[str, str]:
"""Look up fictional order ORD-104; this demo has no real customer data."""
try:
return lookup_order(order_id)
except ValueError as error:
raise ToolError(str(error)) from error
if __name__ == "__main__":
server.run(transport="stdio")
SDK 會根據函式簽章與 docstring 產生工具定義。伺服器使用 stdio，由用戶端將其作為子行程啟動，並透過標準串流交換協定訊息。請勿在伺服器的標準輸出上寫入應用程式日誌。
執行：
python agent.py --mode mcp
main() 中的 mcp 分支會啟動 server.py、呼叫 list_tools()，並檢查訂單工具的契約。當模型請求查詢時，相同的分派器現在改用 client.call_tool()。伺服器執行該函式，應用程式再將結果回傳給模型。
以相同準則比較兩次執行：正確的工具、引數為 ORD-104、成功結果包含 shipped，以及與結果一致的最終答案。執行路徑改變，但任務與面向模型的函式保持一致。
以上三個原始檔皆已包含於本文。示例套件亦包含它們及離線檢查，方便重用。這些檢查會以腳本化的模型回應對本機 MCP 伺服器進行實測；線上模型輸出與該本地驗證分離。
把固定的訂單資料換成你的應用實際需要的最小操作。對於運送查詢，回傳訂單 ID 與有用的狀態欄位。精簡的結果能減少模型需要解讀的無關資訊，並讓答案更易於驗證。
保持工具的描述與引數契約具體明確。此範例只接受一個字串，且宿主在分派前會先檢查請求的工具名稱。MCP 的安全建議要求伺服器驗證輸入並實施存取控制。對於真實的訂單服務，請從可信的應用程式情境推導客戶身分，並在服務內強制限制其可存取的訂單。模型提供的訂單 ID 並不能證明所有權。
當工具可用後，請評估已完成的任務、端對端時間與實際成本。當 MCP 的標準服務介面對需要該工具的應用程式有幫助時就加入 MCP；對於較易運維的情況，則維持直接處理器。
兩者處理的介面不同：MCP 連接應用程式的用戶端與伺服器。在此範例中，模型請求一個函式，而應用程式透過 MCP 用戶端完成該請求。兩者一同參與同一個代理迴圈。
可以。將 MCP 工具內的查詢替換為你的 API 呼叫即可。MCP 伺服器發布工具契約並處理工具呼叫；你的既有服務持續提供其商務行為。
當用戶端需要連線到獨立部署的伺服器，而非啟動本機行程時。MCP 為該連線定義了 Streamable HTTP，也支援作為用戶端啟動的 stdio 子行程。請為遠端伺服器配置其目標呼叫方所需的授權。
對單一應用程式而言，先將 orders.py 中的虛構紀錄替換為你服務中的一個唯讀查詢，並以直接模式開始。
若你需要既有的 MCP 服務或共享工具介面，則採用 MCP 路徑，同時保持相同檢查：正確的操作、正確的資料紀錄，以及由回傳資料支持的答案。