Xây dựng với CometAPICần một API cho nhiều mô hình?Kiểm thử quy trình trong Playground, sau đó tạo API key khi mô hình và chi phí phù hợp với trường hợp sử dụng của bạn.
MCP so với gọi hàm: Cách kết nối công cụ với tác nhân AI
So sánh MCP với gọi hàm và xây dựng một vòng lặp công cụ hoàn chỉnh với CometAPI. Theo mã Python để chạy cùng một tra cứu đơn hàng trực tiếp và thông qua MCP.
Gọi hàm cung cấp cho mô hình cách để yêu cầu một công cụ. MCP cung cấp cho ứng dụng một cách chuẩn để khám phá và gọi các công cụ trên máy chủ. Dùng tích hợp hàm trực tiếp khi bạn có một bộ công cụ nhỏ do ứng dụng sở hữu. Thêm MCP khi bạn muốn tiêu thụ một dịch vụ tương thích hiện có hoặc phơi bày các công cụ qua một giao diện dùng chung. Bên dưới, bạn sẽ kết nối cùng một tra cứu đơn hàng theo cả hai cách qua CometAPI, sử dụng một vòng lặp tác nhân hoàn chỉnh.
Key Takeaways
Bắt đầu với phân phối trực tiếp khi vài công cụ phục vụ một ứng dụng; dùng MCP khi một dịch vụ công cụ tương thích hoặc một giao diện công cụ dùng chung giúp việc tích hợp dễ bảo trì hơn.
Cả hai hướng đều cần một vòng lặp hoàn chỉnh: mô tả công cụ, nhận yêu cầu gọi, thực thi, trả kết quả và để mô hình trả lời.
Đã được xem xét về độ rõ ràng, ghi nguồn và thuật ngữ API hiện tại.
Bạn có thể giữ nguyên hàm đối mặt với mô hình trong khi thay đổi cách thực thi từ bộ xử lý trực tiếp sang client MCP.
Giữ hợp đồng công cụ hẹp và kiểm tra kết quả so với nhiệm vụ bạn muốn tác nhân hoàn thành.
How Function Calling and MCP Fit Together
Một trợ lý hỗ trợ có thể giải thích việc giao hàng chậm, nhưng nó cần truy cập hệ thống đơn hàng để cho khách biết kiện hàng đang ở đâu. Gọi hàm cung cấp cho mô hình cách yêu cầu tra cứu đó. Ứng dụng của bạn thực thi tra cứu và trả kết quả. Bộ xử lý có thể dùng thư viện cục bộ, cơ sở dữ liệu hoặc API từ xa.
Để đáp ứng yêu cầu đó qua MCP, ứng dụng dùng một client để khám phá các công cụ máy chủ với tools/list và gọi chúng bằng tools/call. Ứng dụng vẫn quyết định công cụ nào trong số đã khám phá sẽ được phơi bày cho mô hình của mình.
Cho phép mô hình yêu cầu một thao tác được đặt tên với đối số có cấu trúc
Chuẩn hóa cách các ứng dụng tương thích khám phá và gọi các công cụ của máy chủ
Định nghĩa công cụ
Định nghĩa được cung cấp theo định dạng API của mô hình
Định nghĩa do máy chủ cung cấp, bao gồm inputSchema
Vị trí thực thi
Trình xử lý mà ứng dụng phân phối tới, có thể gọi dịch vụ cục bộ hoặc từ xa
Triển khai công cụ trên máy chủ MCP, có thể gọi dịch vụ backend
Có cần máy chủ riêng?
Bản thân giao diện gọi hàm không yêu cầu máy chủ
Cần máy chủ MCP, nhưng client có thể khởi chạy nó như một tiến trình con stdio cục bộ; dịch vụ từ xa triển khai riêng là tùy chọn
Trường hợp điển hình
Vài công cụ thuộc sở hữu ứng dụng với phân phối trực tiếp
Các dịch vụ tương thích hiện có hoặc một giao diện công cụ được tái sử dụng giữa nhiều ứng dụng
Có thể hoạt động cùng nhau?
Ứng dụng có thể định tuyến yêu cầu hàm của mô hình qua MCP
Client MCP đáp ứng yêu cầu đó bằng tools/call và trả kết quả về cho ứng dụng
Hàng cuối cùng là chìa khóa để xây dựng với cả hai. Mô hình có thể yêu cầu get_order_status, trong khi ứng dụng chọn gọi bộ xử lý riêng hoặc một máy chủ MCP. Định nghĩa hàm đối mặt với mô hình có thể giữ nguyên.
Choose the integration your application needs
Chúng tôi khuyến nghị bắt đầu với tích hợp trực tiếp khi bạn sở hữu vài công cụ dùng bởi một ứng dụng. Ít thứ phải triển khai hơn, và đường đi từ yêu cầu của mô hình tới mã nghiệp vụ của bạn dễ theo dõi.
MCP trở nên hữu ích khi một máy chủ tương thích đã cung cấp năng lực, hoặc khi nhiều ứng dụng AI cần một giao diện công cụ chung. Cùng một công cụ đơn hàng có thể phục vụ trợ lý hỗ trợ và ứng dụng vận hành nội bộ, với mỗi host kiểm soát cách người dùng của họ truy cập nó.
Một dispatcher có thể hỗ trợ cả hàm nội bộ và dịch vụ MCP, định tuyến mỗi thao tác tới triển khai của nó. Thư viện và REST API cũng có thể dùng chung; chọn MCP khi giao diện khám phá và gọi tiêu chuẩn của nó hữu ích cho các ứng dụng tiêu thụ công cụ.
Build a complete tool loop through CometAPI
Chúng ta sẽ dùng một thao tác chỉ-đọc, get_order_status, và một đơn hàng hư cấu, ORD-104. Ví dụ đầu tiên gọi thao tác trực tiếp, sau đó chạy nó qua một máy chủ MCP cục bộ. Cả hai chế độ đều dùng CometAPI Responses API cho suy luận mô hình.
Hình 2. Một vòng lặp tác nhân có thể hỗ trợ thực thi trực tiếp và MCP. Minh họa gốc, dựa trên kiến trúc MCP và luồng gọi hàm của CometAPI.
Step 1 Set up the project
Dùng Python 3.11 trở lên và khóa API CometAPI. Hướng dẫn này được kiểm thử với Python 3.12.14 và mcp==2.3.0. Làm theo CometAPI quickstart để tạo khóa. Trên shell macOS hoặc Linux, tạo dự án và kích hoạt môi trường ảo:
Thay thế placeholder bằng khóa của bạn và giữ nó ngoài mã nguồn công khai. Hướng dẫn này cố định phiên bản MCP SDK dùng trong ví dụ; hướng dẫn máy chủ Python chính thức nhắm tới API 2.x. Hai lệnh tác nhân dưới đây gửi yêu cầu tới CometAPI và tiêu tốn tín dụng API.
Step 2 Define the order lookup
Lưu đoạn này thành orders.py. Dữ liệu cố định giữ cho hướng dẫn tự chứa; sau này bạn có thể thay tra cứu bằng dịch vụ của riêng bạn.
"""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"}
Gọi get_order_status("ORD-104") trả về fixture này:
Lưu script hoàn chỉnh sau thành agent.py. Nó đăng ký hàm, gửi câu hỏi tới CometAPI, thực thi các yêu cầu gọi và trả mỗi kết quả bằng call_id gốc.
Định nghĩa đối mặt với mô hình dùng trường parameters của CometAPI Responses API. MCP cung cấp inputSchema; ví dụ này kiểm tra hợp đồng đã khám phá so với schema một chuỗi order_id trước khi phơi bày hàm đã rà soát.
"""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())
Vòng lặp giữ nguyên câu hỏi ban đầu và mọi mục phản hồi được trả về. Theo định dạng hội thoại CometAPI Responses, nó xóa các trường id cấp cao của mục đầu ra trước khi phát lại và thêm một function_call_output cho mỗi lần gọi. Việc giữ lại các mục còn lại cũng giữ bất kỳ mục lý luận nào cần cho lượt tiếp theo.
Step 4 Run the direct path
Chạy:
python agent.py --mode direct
Ứng dụng mô tả get_order_status cho mô hình. Khi mô hình yêu cầu nó, dispatch() kiểm tra tên và đối số, rồi gọi hàm trong orders.py. Kết quả được đưa trở lại cuộc hội thoại, và mô hình có thể chuyển nó thành câu trả lời.
Dùng dấu vết công cụ in ra terminal của bạn để kiểm tra rằng tra cứu dùng ORD-104 và trả status: shipped. Câu trả lời cuối cùng nên phản ánh kết quả đó; cách diễn đạt có thể khác nhau. Giờ bạn đã có vòng lặp gọi hàm hoàn chỉnh, từ câu hỏi của người dùng tới thao tác do ứng dụng thực thi và phản hồi tiếp theo của mô hình.
Bước tiếp theo giữ nguyên mô hình và vòng lặp hội thoại. Nó chuyển thực thi ra phía sau một máy chủ MCP và cho phép ứng dụng khám phá và gọi công cụ thông qua client.
Step 5 Expose the lookup as an MCP tool
Lưu đoạn này thành server.py, bên cạnh hai tệp còn lại:
"""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 xây dựng định nghĩa công cụ từ chữ ký hàm và docstring. Máy chủ dùng stdio, nơi client khởi chạy nó như một tiến trình con và trao đổi thông điệp giao thức qua các luồng chuẩn của nó. Giữ log ứng dụng khỏi stdout của máy chủ.
Step 6 Run the MCP path
Chạy:
python agent.py --mode mcp
Nhánh mcp trong main() khởi chạy server.py, gọi list_tools() và kiểm tra hợp đồng của công cụ đơn hàng. Khi mô hình yêu cầu tra cứu, cùng dispatcher giờ dùng client.call_tool(). Máy chủ chạy hàm, và ứng dụng trả kết quả của nó về cho mô hình.
So sánh hai lần chạy theo cùng tiêu chí: công cụ đúng, ORD-104 là đối số, kết quả thành công chứa shipped, và câu trả lời cuối cùng nhất quán với kết quả đó. Đường thực thi đã thay đổi; nhiệm vụ và hàm đối mặt với mô hình vẫn giữ nguyên.
Cả ba tệp nguồn đều được bao gồm ở trên. Gói ví dụ cũng chứa chúng và các kiểm tra offline để tái sử dụng tiện lợi. Những kiểm tra đó chạy một máy chủ MCP cục bộ thực với phản hồi mô hình kịch bản; đầu ra mô hình trực tiếp tách biệt với xác minh cục bộ đó.
Adapt the example to your own tools
Thay dữ liệu đơn hàng cố định bằng thao tác nhỏ nhất mà ứng dụng của bạn thực sự cần. Với câu hỏi giao hàng, trả về ID đơn hàng và các trường trạng thái hữu ích. Kết quả gọn nhẹ cho mô hình ít vật liệu không liên quan để diễn giải và khiến câu trả lời dễ kiểm tra hơn.
Giữ mô tả công cụ và hợp đồng đối số cụ thể. Ví dụ này chấp nhận đúng một chuỗi, và host kiểm tra tên công cụ được yêu cầu trước khi phân phối. Hướng dẫn bảo mật MCP kêu gọi máy chủ xác thực đầu vào và triển khai kiểm soát truy cập. Với dịch vụ đơn hàng thực, lấy danh tính khách hàng từ bối cảnh ứng dụng đáng tin cậy và thực thi quyền truy cập vào đơn hàng của khách hàng đó bên trong dịch vụ. Một ID đơn hàng do mô hình cung cấp không thể thiết lập quyền sở hữu.
Khi công cụ hoạt động, đánh giá nhiệm vụ đã hoàn thành, thời gian end-to-end và chi phí thực tế. Thêm MCP khi giao diện dịch vụ của nó giúp các ứng dụng cần công cụ, và giữ các bộ xử lý trực tiếp ở nơi chúng đơn giản hơn để vận hành.
FAQ
MCP có thay thế gọi hàm không?
Chúng xử lý các giao diện khác nhau: MCP kết nối client của ứng dụng tới máy chủ. Trong ví dụ này, mô hình yêu cầu một hàm và ứng dụng đáp ứng yêu cầu đó qua client MCP. Cả hai cùng tham gia vào một vòng lặp tác nhân.
Một công cụ MCP có thể gọi REST API hiện có không?
Có. Thay tra cứu bên trong công cụ MCP bằng một cuộc gọi tới API của bạn. Máy chủ MCP công bố hợp đồng công cụ và xử lý các lần gọi công cụ; dịch vụ hiện có của bạn tiếp tục cung cấp hành vi nghiệp vụ.
Khi nào tôi nên dùng máy chủ MCP từ xa?
Dùng dịch vụ từ xa khi các client cần truy cập một máy chủ triển khai riêng thay vì khởi chạy tiến trình cục bộ. MCP định nghĩa Streamable HTTP cho kết nối đó và stdio cho tiến trình con do client khởi chạy. Cấu hình ủy quyền của máy chủ từ xa cho các caller dự định.
Conclusion
Với một ứng dụng đơn lẻ, thay bản ghi hư cấu trong orders.py bằng một truy vấn chỉ-đọc từ dịch vụ của bạn và bắt đầu với chế độ trực tiếp.
Nếu bạn cần một dịch vụ MCP hiện có hoặc một giao diện công cụ dùng chung, dùng hướng MCP và giữ cùng các kiểm tra: thao tác đúng, bản ghi đúng và một câu trả lời được hỗ trợ bởi dữ liệu trả về.