大模型終於能“聽懂”雲操作了？

2025-06-25 12:33 阿里雲開發者

阿里妹導讀
本文透過 MCP Server 和大模型的結合，實現雲產品管理的自然語言操作，極大提升開發者的操作效率和使用者體驗。
前言
對於阿里雲使用者而言，管理雲產品通常有幾種方式：控制檯、OpenAPI SDK、CADT、Terraform 等。雖然形式不同，但本質上這些操作都要依賴於 OpenAPI 來完成。在 MCP 出現之前，如果使用者希望透過自然語言來操作雲產品，唯一的辦法就是對照產品文件，手動編寫一條條 FunctionCall 描述，再交給大模型進行呼叫。這種方式不僅繁瑣，而且效率低下。幾周前，當得知 OpenAPI 推出內測版的 MCP Server 時，我意識到一個重要的機會來了。設想一下，在簡單配置好 MCP Server 後，只需問一句：“我在杭州可用區有哪些資源？”、“哪個Region有L20 顯示卡，費用是多少？”、“杭州某個 ACK 叢集當前的負載情況如何？”等問題，大模型就能自動呼叫相關工具，然後給出準確的回答。這種互動體驗和效率提升將是前所未有的。在這篇小文中，我會結合自己的實際使用情況，和大家一起“趟趟水”、“踩踩坑”。本文主要基於個人的理解與實踐，只講大白話，如有理解錯誤的地方非常歡迎大家批評、指正。
一、上手體驗
以下以CherryStudio為例，簡要介紹一下快速上手過程。
1.1 如何配置
總共分三步：
第一步：點擊OpenAPI MCP[1]服務來建立一個雲產品的MCP Server，每個雲產品最多選30個；

圖1. 選擇某雲產品（如ECS）API列表
第二步：確保已登入阿里雲賬號，CherryStudio需要透過OAuth2.0來授權，如果使用的是RAM賬號的話需要提前新增服務策略；
第三步：配置CherrySudio，該軟體是一款國人開發的多模型聚合、一體化工具，有很多現成的智慧體（主要是提示詞），也和百鍊MCP做了深度整合。

圖2. 在CherryStudio裡配置MCP Server
當右上角的按鈕變成綠色，整個配置就結束了。看，就是這麼快。
備註：1、CherryStudio需要大模型API Key，建議到百鍊申請；2、第一次使用會跳出一個授權對話方塊，選擇同意即可。
1.2 如何使用
我們選擇大模型為“qwen-max"，選上MCP服務，在對話方塊裡詢問LLM關於雲產品相關問題，這裡可以看到LLM的呼叫工具情況。

圖3. 在CherryStudio查詢雲產品情況
1.3 有些什麼問題
使用中還是不少問題的，比如要頻繁選擇不同MCP Server、大模型響應慢（qwen3-235b-a22b、max使用效果比較好），其中最主要的問題是下面這個：

圖4. 由於超出大模型上下文視窗長度限制而報錯
為了解決這個問題，我們只能在設定裡手工disable此次詢問可能不需要的工具，之後大模型又能正確工作了。

圖5. 設定中取消部分API以減少上下文長度
大家看到這裡是不是都會心裡涼了一把，“這也太麻煩了，我還得提前根據客戶問什麼再看選什麼API列表，這麼折騰誰用啊”。實話說，我當時也是這麼想的。到了這裡我覺得更有必要去細緻研究一下，探究內部原理到底是啥。
提前說一句，本文第三章給出一種通用解決方案、可以有效解決該問題，請大家繼續往下看:-)。
二、原理探究
這裡以Qwen-max、Agno（本文稱為LLM框架，也是MCP Host）、ECS OpenAPI MCP Server為例進行梳理。
Agno框架已有27.6K star（對比LangGraph 13.4k），不限所用模型，執行速度快。
2.1 基礎背景
以下四條是理解原理的基礎背景，我們會以這些原則來理解後續一系列動作。
1、大模型是無狀態、無記憶的：除非有內部偏好或記憶設定，各家大模型基本上都是無狀態的，也只有這樣才能撐起高併發；
2、關注同一個會話裡的上下文：在瀏覽器的對話窗口裡，每次會話都會把對話歷史資訊再發給大模型（根據模型上下文長度限制進行滑動擷取），原因也是第一條。在框架裡為了靈活控制，一般由使用者來選擇是否把歷史資訊發給大模型；
3、框架會做很多事：為了方便使用者使用，框架會做很多事情，有些暴露給客戶，有些則內部處理而不通知客戶，如果不看原始碼和幫助文件，框架的行為將是一個黑盒；
4、Agent就是“專家”：只要是圍繞著LLM組織起來、具備某種專項能力的，就是Agent，也可以理解為專家。Agent的實現形式，可以是簡單地定製一個SystemPrompt，也可以是使用WorkFlow、Memory、RAG等複雜元件來實現。從這個角度看，框架（LangChain、Agno等）也是AI Agent。
2.2 整體框架
這裡的Host可以是Agno框架，也可以是CherryStudio應用。

圖6. MCP整體框架
簡要說明：
1）Host會採用1:1的方式建立MCP Client，與遠端MCP Server建連，可選SSE或Streamable http兩種方式，建議優選Streamable http，因為其效能高、資源消耗低；
2）Host透過MCP協議拿到所有Tools（也就是API），傳送給LLM供選擇；
3）LLM根據使用者提問選擇相關工具，將資訊發給Host，Host呼叫MCP服務後返回響應；
4）當所需工具呼叫完，LLM會綜合分析結果，然後由Host反饋給到使用者。
透過上圖我們可以看到，MCP Server基本上起一個代理的做用，也可以看做是MCP協議的解除安裝點。因此，如果MCP Server本身比較輕量、主要由後端提供服務的話，它完全可以做得很輕，甚至整合到AI閘道器裡。
需要說明的是，OpenAPI MCP Server都是遠端連線的，而MCP Client本身也支援本地連線（採用STDIO模式），這種情況下MCP Server執行在本地，一般會執行Node.js（使用npx命令，下載到本地後執行）或Python（使用uv命令，也是下載到本地後執行）。具體可參考本文文末的參考資料。
2.3 互動流程
這裡以Agno的視角來看具體互動流程，其他框架類似。

圖7. MCP互動流程
簡要說明：
1）Agno根據MCP Server個數建立相應個數的MCP Client，使用JsonRPC方式與MCP Server互動；
2）初始化完成後，透過tools/list獲得Server提供的所有工具列表；
3）當客戶發來查詢請求時，Host將所有工具資訊放在SystemPrompt、連同使用者請求一起發給LLM；
4）LLM判斷需要呼叫工具時，傳送<tool_use>給Agno，Agno據此呼叫tools/call來返回響應；
5）整個互動過程中，根據使用者配置來顯示呼叫資訊，返回最終的LLM輸出資訊。
2.4 MCP Inspector
MCP官網提供了一個Node工具，可以很方便地本地執行這個工具來對MCP Server進行驗證。
  • 使用方式：本地視窗執行“npx @modelcontextprotocol/inspector”，該命令會從遠端拉取服務並在本地執行，透過瀏覽器開啟“Http://127.0.0.1:6274”來使用。
  • 典型使用：

圖8. Inspector配置

圖9. Initialize和tools/list命令及響應
三、程式碼驗證
以下程式碼均在本地Mac中除錯透過，如需移植到其他平臺（如ECS）需要提工單申請配置OAuth CallBack路徑。
3.1 OAuth2.0授權
首先，按照OAuth標準的“Metadata Discovery”方式，透過在MCP Server地址拼接".well-known/oauth-authorization-server"來獲得OAuth伺服器資訊：

圖10. OAuth伺服器元資料資訊
接著，使用PKCE方式來獲取AccessToken，注意這裡的ClientID可使用預先定義的ID，若無則需重新註冊一下：
from utility import set_keyapp = Flask(__name__)app.secret_key = secrets.token_urlsafe(16)CLIENT_ID = "40711518457*******"REDIRECT_URI = "http://127.0.0.1:5000/oauth/callback"DISCOVERY_URL = "https://openapi-mcp.cn-hangzhou.aliyuncs.com/.well-known/oauth-authorization-server"deffetch_discovery_info():"""從 discovery url 獲取 Oauth 端點資訊"""try:        resp = requests.get(DISCOVERY_URL, timeout=5)if resp.status_code == 200:            data = resp.json()return {"authorization_endpoint": data.get("authorization_endpoint"),"token_endpoint": data.get("token_endpoint"),"registration_endpoint": data.get("registration_endpoint")            }except Exception as e:print(f"Failed to fetch discovery info: {e}")return {}# 預設端點AUTHORIZATION_ENDPOINT = "https://signin.aliyun.com/oauth2/v1/auth"TOKEN_ENDPOINT = "https://oauth.aliyun.com/v1/token"defgenerate_pkce():"""生成 PKCE 的 code_verifier 和 code_challenge"""    code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=")# 計算 S256 code_challenge    digest = hashlib.sha256(code_verifier.encode()).digest()    code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")return code_verifier, code_challenge@app.route("/")defhome():return'<a href="/login">Login with OAuth</a>'@app.route("/login")deflogin():    registration_endpoint = ""# 嘗試用 discovery 資訊覆蓋端點    discovery = fetch_discovery_info()print(f"Discovery info: {discovery}")if discovery.get("authorization_endpoint"):        AUTHORIZATION_ENDPOINT = discovery["authorization_endpoint"]if discovery.get("token_endpoint"):        TOKEN_ENDPOINT = discovery["token_endpoint"]if discovery.get("registration_endpoint"):        registration_endpoint = discovery["registration_endpoint"]# 註冊一個 client（如果 CLIENT_ID 未設定或為佔位符）    client_id = CLIENT_IDif (not client_id) or client_id.endswith("*******"):        ifnot registration_endpoint:return"Registration endpoint not available", 400# 註冊 client        reg_data = {"redirect_uris": [REDIRECT_URI],"grant_types": ["authorization_code"],"response_types": ["code"],        }try:            reg_resp = requests.post(registration_endpoint, json=reg_data, timeout=5)if reg_resp.status_code != 201:returnf"Client registration failed: {reg_resp.text}", 400            reg_json = reg_resp.json()            client_id = reg_json.get("client_id")            ifnot client_id:return"No client_id returned from registration", 400            session["client_id"] = client_idexcept Exception as e:returnf"Client registration exception: {e}", 400else:        session["client_id"] = client_id# 生成 PKCE 引數    code_verifier, code_challenge = generate_pkce()# 生成隨機 state 防止 CSRF    state = secrets.token_urlsafe(16)# 儲存到 session    session.update({"code_verifier": code_verifier,"state": state    })# 構造授權請求 URL    params = {"response_type": "code","client_id": session["client_id"],"redirect_uri": REDIRECT_URI,"code_challenge": code_challenge,"code_challenge_method": "S256","state": state    }    auth_url = f"{AUTHORIZATION_ENDPOINT}?{urllib.parse.urlencode(params)}"return redirect(auth_url)@app.route("/oauth/callback")defcallback():# 檢查錯誤響應if"error"in request.args:returnf"Error: {request.args['error']}"# 驗證 stateif request.args.get("state") != session.get("state"):return"Invalid state parameter", 400# 獲取授權碼    auth_code = request.args.get("amp;code") or request.args.get('code')  # 嘗試兩種可能的引數名    ifnot auth_code:return"Missing authorization code", 400# 用授權碼換取 token    data = {"grant_type": "authorization_code","code": auth_code,"redirect_uri": REDIRECT_URI,"client_id": session.get("client_id", CLIENT_ID),"code_verifier": session["code_verifier"]    }    response = requests.post(TOKEN_ENDPOINT, data=data)if response.status_code != 200:returnf"Token request failed: {response.text}", 400    token_info = response.json().get("access_token")# 儲存到本地配置檔案print(f"Your access_token: {token_info}")    set_key("ALI_OPENAPI_ACCESS_TOKEN", token_info)# 刪掉session引數    session.pop("code_verifier", None)    session.pop("state", None)return response.json()if __name__ == "__main__":    app.run(port=5000, debug=True)
最後，在已經登入阿里雲賬號的狀態下使用瀏覽器訪問“http://127.0.0.1:5000”來獲取access_token，將此token存入到本地配置檔案中。與1.1節裡CherryStudio在彈出的介面裡要求授權一樣，兩者的目的都是為獲得access_token，這個token將在下次訪問MCP Server時配置在Heard中的Authorization中，其格式為{'Authorization': f'Bearer {access_token}'}。類似的，在2.4節中的MCP Inspector中也要配置此項。
說明：1）access_token預設有效時長259199s（3天），在失效前建議使用refresh_token來請求更新access_token；2）官方Python SDK的MCP Client也支援OAuth鑑權，讀者可參照說明自行實現。
3.2 第一個MCP應用
拿到access_token後，我們就可以實現這個應用：
print(f"Current path is {os.getcwd()}")load_keys()async defrun_agent(message: str) -> None:    server_params = StreamableHTTPClientParams(        url = "https://openapi-mcp.cn-hangzhou.aliyuncs.com/accounts/1411741061209533/custom/ecs_tst_agno/id/RXPfhaBVHq7w3wkp/mcp,    # ECS        headers = {'Authorization': f'Bearer {os.getenv("ALI_OPENAPI_ACCESS_TOKEN")}'}    )    async with MCPTools(server_params=server_params, transport="streamable-http", timeout_seconds=30 ) as mcp_tools:        # Initialize the model        model=OpenAILike(id="qwen-max",                          api_key=os.getenv("DASHSCOPE_API_KEY"),                         base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")        # Initialize the agent        agent = Agent(model=model,                      tools=[mcp_tools],                      instructions=dedent("""\                          你是一個阿里云云計算專家，請根據使用者的問題，使用MCP服務查詢阿里雲的雲產品資訊，給出詳細的解釋。                          請使用中文回答"""),                      markdown=True,                      show_tool_calls=True)        # Run the agent        await agent.aprint_response(message, stream=True)# Example usageif __name__ == "__main__":    asyncio.run(run_agent("我在上海有哪些ECS例項?"))
程式執行結果如下：

圖11. MCP應用執行結果
說明：1）Agno框架的mcp.py的174行有bug，先使用“client_timeout = self.timeout_seconds”來跳過；2）執行時會報錯“JSONRPCRequest.method. Field required [type=missing, input_value={'result': 'ok'}”，這個主要與Server端實現有關，內部已排期修復。不影響正常使用。
3.3 解決"Input length ouf of range"
1.3節中我們使用的ECS MCP Server有26個工具，在使用全量工具集、模型為Qwen-max的時候CherryStudio會報“超出上下文長度限制”的錯誤。在Agno中，報錯類似：

圖12. Agno報錯上下文長度超限
由2.3的互動流程可以發現，上下文超標主要是因為tools集合資料太多導致。進一步觀察後，發現每個tool由name、description、inputSchema、annotaions四部分組成，其中inputSchema的佔用量最大。為此，我們可以讓LLM先根據全集列表中的name、description判斷一把，告訴我們需要哪些tool，然後再配置LLM，這樣便能有效減少上下文長度。具體流程如下：

圖13. 預篩選-呼叫流程
程式碼實現如下：
1）使用mcp client 獲取全量tools資訊，然後由LLM根據使用者請求判斷要哪些工具。注意這裡的提示詞裡要明確說明不能返回"“`"、"json"等字串，不然無法進行格式化。
defget_selected_tools_list(server_url, hearders, llm_api_key, user_question):from mcp.client.streamable_http import streamablehttp_clientfrom mcp import ClientSessionimport asynciofrom agno.agent import Agent, RunResponsefrom agno.models.openai.like import OpenAILikeimport json# Important: Just to avoid such logging error like "JSONRPCError.jsonrpc Field required ...import logging    logging.disable(logging.CRITICAL)# 1. Get all tools via tools/list    all_tools = Noneasyncdefget_all_tools():# Connect to a streamable HTTP serverasyncwith streamablehttp_client(url=server_url,headers=hearders)as(read_stream, write_stream,_):# Create a session using the client streamsasyncwith ClientSession(read_stream, write_stream) as session:await session.initialize()                all_tools = await session.list_tools()print(f"Number of tools: {len(all_tools.tools)}")return all_tools    all_tools = asyncio.run(get_all_tools())# 2. Collect all tools brief info    brife_tools_info = [        {"name": tool.name,"description": tool.description,        }for tool in all_tools.tools]# 3. Create an agent    simple_agent = Agent(        model=OpenAILike(id="qwen-max",            api_key=llm_api_key,            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"        ),        system_message="""            你是一個雲計算專家，嚴格按照請客戶提供的上下文資訊回答問題。            """,        )# 4. Run the agent to tell us which tools are needed for user questio    prompt = (f"請根據提供的工具資訊以及使用者請求，你需要給出可能需要呼叫的api列表，以Json形式返回，要精簡、不需要其他資訊。格式上，返回值不帶```、json等字串，"f"工具資訊如下：{json.dumps(brife_tools_info)},"f"現在客戶提問：{user_question}"    )    response: RunResponse = simple_agent.run(prompt)print(response.content)    tool_to_use_list = json.loads(response.content)print(f"Selected Tool List: {tool_to_use_list}")return tool_to_use_list
2）MCPTools構造時加入白名單（include_tools=config["tool_to_use_list"]），只有白名單上的tool可以被調取，然後正式發起查詢。
print(f"Current path is {os.getcwd()}")load_keys()asyncdefrun_agent(config):# Setup agent with MCP tools    server_params = StreamableHTTPClientParams(url=config["server_url"], headers=config["headers"])asyncwith MCPTools(server_params=server_params,                         transport="streamable-http",                         timeout_seconds=30,                         include_tools=config["tool_to_use_list"]) as mcp_tools:# Initialize the model        model=OpenAILike(id="qwen-max",                          api_key=config["llm_api_key"],                         base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# Initialize the agent        agent = Agent(model=model,                      tools=[mcp_tools],                      instructions=dedent("""\                          你是一個阿里云云計算專家，請根據使用者的問題，使用MCP服務查詢阿里雲的雲產品資訊，給出詳細的解釋。                          請使用中文回答                      """),                      markdown=True,                      show_tool_calls=True)# Run the agentawait agent.aprint_response(config["user_question"], stream=True)# Example usageif __name__ == "__main__":# 1. User question goes here    user_question = "我在上海有哪些ECS例項?"# 2. Prepare arguments for the agent    url = "https://openapi-mcp.cn-hangzhou.aliyuncs.com/accounts/1411741061209533/custom/ecs_mcp/id/vwAaigJjvaOkaqHf/mcp"# Full ECS List    headers = {'Authorization': f'Bearer {os.getenv("ALI_OPENAPI_ACCESS_TOKEN")}'}    llm_api_key = os.getenv("DASHSCOPE_API_KEY")# 3. Get selected tools list by user question    seleted_tools = get_selected_tools_list(url, headers, llm_api_key, user_question)# print(seleted_tools)# # 4. Setup config    config = {"server_url": url,"llm_api_key": llm_api_key,"headers": headers,"user_question": user_question,"tool_to_use_list": seleted_tools    }# 5. Query LLM    asyncio.run(run_agent(config))
當用戶詢問“我在上海有哪些ECS例項?”時，LLM判斷只要一個工具（Ecs-20140526-DescribeInstances）就夠了，然後再次裝載MCPTools後程序執行正常，結果如下：

圖14. 從26個tools中挑選1個來查詢
3.4 後續最佳化
以上三個小實驗只是從程式碼層面驗證了MCP的基本功能，離真正的工程生產還相差甚遠，後續將在以下方面最佳化：
1）將每個雲產品的MCP Server按照不同維度拆分，比如按照操作型別分為Describe*、Get*、Modify*、Create/Run*等，建立不同種類的MCP Server。這樣可以做到更細粒度的拆分，從源頭減少Tools數量問題；
2）使用MultiMCPTools來組裝多個MCP Server，使LLM可以同時使用多個Server；
3）進一步最佳化工具篩選，根據客戶請求分析多個MCP Server，然後給出篩選建議；
4）使用圖形化介面，將篩選、執行組裝成一個工作流，白屏化操作；
5）對非查詢類（如Create、Delete、Update）的MCP需要引入客戶互動，告知客戶風險，由客戶確認每一步是否可以操作。
四、未來展望
開放平臺現有2W多個OpenAPI，在MCP Server的幫助下這些API可以被更加靈活地利用起來，相信有非常多的實用場景會被創造出來，屆時雲對普通大眾來說也許就是“一兩句話的事兒”。
在此從個人角度地展望一下：
1、推動文件完善：LLM能否有效呼叫tool的關鍵在於文件（描述、引數說明等），只有表達更清楚、更全面才能更有利於LLM決策。這方面開放平臺也允許使用者自行調優；
2、控制幻覺：LLM一大優勢是能幫我們補充各種細節，選工具、填引數、分析結果，可另一方面它也有幻覺，可以生成很多貌似正確的東西，且不同的模型幻覺程度也不一樣，因此必須由人做最終控制和決策。當然，也可以加入另一個LLM來稽核前一個LLM的動作，減輕人的工作量；
3、建立每個雲產品Agent：ECS、NAS、RocketMQ等都可以建立自己的專業Agent（搭配RAG），使用者只要描述需求，剩下將由專業Agent在最佳實踐基礎上一鍵搞定；
4、搭建Agent矩陣：透過A2A協議在不同的雲產品間進行協作，為客戶提供完整解決方案；
5、實現高度安全：制定整體安全框架，確保Agent所有行為安全、可控。

附：參考資料

1、阿里雲官網 – OpenAPI MCP控制檯：https://api.aliyun.com/mcp
2、Model Context Protocol：https://modelcontextprotocol.io/docs/concepts/architecture
3、MCP – Python SDK：https://github.com/modelcontextprotocol/python-sdk
4、Agno – Full-stack framework for building Multi-Agent System：https://docs.agno.com/introduction
通義千問3 + MCP：一切皆有可能
MCP 協議透過標準化互動方式解決 AI 大模型與外部資料來源、工具的整合難題；通義千問3 原生支援 MCP 協議，能更精準呼叫工具；阿里雲百鍊上線了業界首個全生命週期 MCP 服務，大幅降低 Agent 開發門檻，使用者只需 5 分鐘即可構建增強型智慧體。
點選閱讀原文檢視詳情。

相關文章