由于企業(yè)部署Agent時(shí)需要整合的工具被集成為Agent本身內(nèi)部的API調(diào)用，導(dǎo)致擴(kuò)展性和復(fù)用性較差。亞馬遜云科技Amazon SageMaker AI的為其推出的免費(fèi)托管服務(wù)，目前提供2個(gè)月免費(fèi)試用套餐，能夠提供托管LLM的能力。下文將由站長(zhǎng)百科為大家分享如何通過Amazon SageMaker AI+MCP構(gòu)建自動(dòng)化工作流。

亞馬遜云科技官網(wǎng)：點(diǎn)擊直達(dá)（注冊(cè)即可試用SageMaker AI和百余款云服務(wù)）

參考教程：《亞馬遜云科技賬號(hào)注冊(cè)流程圖解》

一、MCP是什么

近期有關(guān)“MCP”的關(guān)鍵詞在業(yè)界內(nèi)掀起了一波“熱潮”，那么MCP到底是什么呢？MCP是由Anthropic公司開發(fā)的一項(xiàng)開源協(xié)議，主要將AI模型連接到大部分?jǐn)?shù)據(jù)源或工具的標(biāo)準(zhǔn)化方式。MCP采用client-server架構(gòu)（如下圖所示），主要幫助開發(fā)者通過輕量級(jí)的MCP Servers來(lái)公開其數(shù)據(jù)，同時(shí)構(gòu)建作為MCP Clients的AI應(yīng)用程序，且客戶端可連接到服務(wù)器。

傳統(tǒng)API通常將各種功能捆綁在一起，導(dǎo)致需要擴(kuò)展時(shí)必須升級(jí)整個(gè)系統(tǒng)，且更新時(shí)面臨系統(tǒng)全面故障的高風(fēng)險(xiǎn)，而且針對(duì)不同應(yīng)用程序管理不同版本的API也變得異常復(fù)雜和繁瑣。盡管微服務(wù)提供了更高的模塊化程度，但它們通常要求對(duì)每個(gè)服務(wù)單獨(dú)進(jìn)行復(fù)雜集成，造成繁瑣的管理開銷。

MCP的標(biāo)準(zhǔn)化client-server架構(gòu)，專為高效且安全的集成而設(shè)計(jì)，克服了上述限制。它提供了一個(gè)實(shí)時(shí)的雙向通信接口，使得AI系統(tǒng)能夠依據(jù)“一次編寫，隨處使用”的理念，無(wú)縫連接各種外部工具、API服務(wù)以及數(shù)據(jù)源。

以下是MCP Servers與MCP Clients的參考架構(gòu)，采用Amazon SageMaker AI作為基礎(chǔ)模型（FMs）和LLM的托管環(huán)境。盡管此架構(gòu)以Amazon SageMaker AI作為推理核心，但它也可快速調(diào)整以支持Amazon Bedrock模型。該解決方案架構(gòu)如下圖所示：

二、Amazon SageMaker AI集成FastMCP

架構(gòu)確定后，對(duì)下圖所示的應(yīng)用程序流程進(jìn)行分析。

就使用模式而言，MCP與工具調(diào)用在邏輯上存在相似之處，但它首先增加了用于發(fā)現(xiàn)可用工具的步驟。

1、客戶端連接到MCP Server，并獲取可用工具的列表。

2、客戶端使用根據(jù)MCP Server上可用工具列表設(shè)計(jì)的提示詞（類型為“用戶”的消息），來(lái)調(diào)用LLM。

3、LLM根據(jù)需要判斷應(yīng)調(diào)用的工具以及調(diào)用次數(shù)，并作出回復(fù)（類型為“助手”的消息）。

4、客戶端要求MCP Server執(zhí)行工具調(diào)用，并將結(jié)果提供給LLM（類型為“用戶”的消息）。

5、此循環(huán)會(huì)不斷迭代，直至得出最終答案，并可將答案返回給用戶。

6、客戶端斷開與MCP Server的連接。

要?jiǎng)?chuàng)建MCP Server，需使用官方的Model Context Protocol Python SDK。以創(chuàng)建僅包含一個(gè)工具的簡(jiǎn)單服務(wù)器為例，該工具將模擬在廣播電臺(tái)搜索播放次數(shù)最多的熱門歌曲，并以Python字典的形式返回結(jié)果。請(qǐng)務(wù)必添加適當(dāng)?shù)奈臋n字符串以及輸入與輸出類型注解，以便服務(wù)器和客戶端均能正確發(fā)現(xiàn)并使用該資源。

from mcp.server.fastmcp import FastMCP

# instantiate an MCP server client
mcp = FastMCP(“Radio Station Server”)

# DEFINE TOOLS
@mcp.tool()
def top_song(sign: str) -> dict:
“””Get the most popular song played on a radio station”””
# In this example, we simulate the return
# but you should replace this with your business logic
return {
“song”: “In the end”,
“author”: “Linkin Park”
}

@mcp.tool()
def …

if __name__ == “__main__”:
# Start the MCP server using stdio/SSE transport
mcp.run(transport=”sse”)

MCP Servers可以在亞馬遜云科技的計(jì)算服務(wù)上運(yùn)行，例如Amazon EC2、Amazon EKS或Amazon Lambda，隨后MCP Servers可用于安全地訪問亞馬遜云科技云中的其他資源，例如VPC中的數(shù)據(jù)庫(kù)或企業(yè)API以及外部資源。例如部署MCP Servers的一種簡(jiǎn)單方法是利用Amazon Lambda對(duì)Docker鏡像的支持，在Lambda函數(shù)或Amazon Fargate上安裝MCP依賴項(xiàng)。

服務(wù)器設(shè)置完成后，將重點(diǎn)轉(zhuǎn)移至MCP Client，通信始于MCP Client使用可流式傳輸?shù)腍TTP，連接到MCP Server。

from mcp import ClientSession
from mcp.client.sse import sse_client

async def connect_to_sse_server(self, server_url: str):
“””Connect to an MCP server running with SSE transport”””
# Store the context managers so they stay alive
self._streams_context = sse_client(url=server_url)
streams = await self._streams_context.__aenter__()

self._session_context = ClientSession(*streams)
self.session: ClientSession = await self._session_context.__aenter__()

# Initialize
await self.session.initialize()

# List available tools to verify connection
print(“Initialized SSE client…”)
print(“Listing tools…”)
response = await self.session.list_tools()
tools = response.tools
print(“\nConnected to server with tools:”, [tool.name for tool in tools])

連接到MCP Server時(shí)，可使用list_tools() API向服務(wù)器請(qǐng)求可用工具的列表。獲取工具列表及其描述后，可為調(diào)用工具定義系統(tǒng)提示詞。

system_message = (
“You are a helpful assistant with access to these tools:\n\n”
f”{tools_description}\n”
“Choose the appropriate tool based on the user’s question. ”
“If no tool is needed, reply directly.\n\n”
“IMPORTANT: When you need to use a tool, you must ONLY respond with ”
“the exact JSON object format below, nothing else:\n”
“{\n”
‘ “tool”: “tool-name”,\n’
‘ “arguments”: {\n’
‘ “argument-name”: “value”\n’
” }\n”
“}\n\n”
“After receiving a tool’s response:\n”
“1. Transform the raw data into a natural, conversational response\n”
“2. Keep responses concise but informative\n”
“3. Focus on the most relevant information\n”
“4. Use appropriate context from the user’s question\n”
“5. Avoid simply repeating the raw data\n\n”
“Please use only the tools that are explicitly defined above.”
)

工具通常使用類似以下示例的.json模式（schema）進(jìn)行定義，該工具名為top_song，功能為獲取廣播電臺(tái)播放次數(shù)最多的熱門歌曲。

{
“name”: “top_song”,
“description”: “Get the most popular song played on a radio station.”,
“parameters”: {
“type”: “object”,
“properties”: {
“sign”: {
“type”: “string”,
“description”: “The call sign for the radio station for which you want the most popular song. Example calls signs are WZPZ and WKRP.”
}
},
“required”: [“sign”]
}
}

系統(tǒng)提示詞配置完成后，可根據(jù)需要多次運(yùn)行聊天循環(huán)。在該過程中，需要在調(diào)用托管的LLM，與調(diào)用由MCP Server所支持的工具之間不斷切換?？墒褂弥T如Amazon SageMaker Boto3、Amazon SageMaker Python SDK，或者諸如LiteLLM及類似庫(kù)等其他第三方庫(kù)，來(lái)實(shí)現(xiàn)這一功能。

messages = [
{“role”: “system”, “content”: system_message},
{“role”: “user”, “content”: “What is the most played song on WZPZ?”}
]

result = sagemaker_client.invoke_endpoint(…)
tool_name, tool_args = parse_tools_from_llm_response(result)
# Identify if there is a tool call in the message received from the LLM
result = await self.session.call_tool(tool_name, tool_args)
# Parse the output from the tool called, then invoke the endpoint again
result = sagemaker_client.invoke_endpoint(…)

托管在AWS的Amazon SageMaker AI上的模型，在其API中并不原生支持函數(shù)調(diào)用功能，因此需要使用正則表達(dá)式或類似方法來(lái)解析響應(yīng)內(nèi)容。

import re, json

def parse_tools_from_llm_response(message: str)->dict:
match = re.search(r'(?s)\{(?:[^{}]|(?:\{[^{}]*\}))*\}’, content)
content = json.loads(match.group(0))
tool_name = content[“tool”]
tool_arguments = content[“arguments”]
return tool_name, tool_arguments

LLM的響應(yīng)中不再包含任何待處理的工具調(diào)用請(qǐng)求時(shí)，即可將響應(yīng)內(nèi)容視為最終答案，并將其返回給用戶。最后記得關(guān)閉數(shù)據(jù)流，以結(jié)束與MCP Server的交互流程。

相關(guān)推薦：

《AI Agent集成MCP實(shí)用教程》

《Amazon SageMaker+LLaVA實(shí)現(xiàn)圖片自動(dòng)化分類》

《Amazon Bedrock+DeepSeek搭建企業(yè)知識(shí)庫(kù)圖文教程》

• 

  廣告合作

• 

  QQ群號(hào)：4114653