AgentCore：AgentCore Python SDK 使用指南

本文介紹使用 Python 調用 AgentCore 模型、MCP、Skill、記憶和憑證，以及通過架構構建 Agent 服務的方法。雲端資源樣本在 AgentCore 託管環境中運行，自訂資源樣本由應用提供串連配置。

前提條件

1. 已開通 AgentCore，並建立 Workspace。具體操作，請參見管理 Workspace。

2. 已根據需要在 Agent 所屬的 Workspace 中準備資源，並確認 Agent 具有相應存取權限。

3. 使用雲端資源的樣本在 AgentCore 託管的應用運行環境中執行。SDK 從運行環境擷取訪問配置，無需在業務代碼中填寫平台存取金鑰。

本文使用以下樣本資源名稱。請替換為您實際建立的資源名稱。

只調用模型時，無需建立 MCP、Skill、MemoryStore 或額外憑證。

安裝 SDK

基礎 SDK 支援 Python 3.10 及以上版本。本文的 LangChain 服務樣本使用 Python 3.11 及以上版本。

pip install alibabacloud-agentcore-sdk==0.1.1

使用 MCP、HTTP 服務、憑證或框架組成時，按需安裝擴充依賴：

pip install "alibabacloud-agentcore-sdk[mcp,server,credentials,langchain]==0.1.1"

Python 的安裝包名為 alibabacloud-agentcore-sdk，代碼中的匯入名稱為 agentcore。

本文樣本使用 Python SDK 0.1.1。後續版本請參見 PyPI。

調用雲端模型

在代碼中指定模型串連名稱和模型名稱。SDK 根據當前 Agent 所屬的 Workspace 尋找模型串連，無需手動拼接模型訪問地址。

import asyncio
from agentcore import AsyncAgentCore

async def main():
    async with AsyncAgentCore.auto() as core:
        model = await core.model("my-model-connection", model="qwen3.8-max")
        response = await model.invoke(
            [{"role": "user", "content": "用一句話介紹杭州。"}],
            temperature=0.2,
        )
        print(response["choices"][0]["message"]["content"])

asyncio.run(main())

async for chunk in model.stream([{"role": "user", "content": "用一句話介紹杭州。"}]):


    for choice in chunk.get("choices", [ ]):


        print(choice.get("delta", {}).get("content") or "", end="", flush=True)

使用 MCP 工具

以下片段在已建立的 core 生命週期內執行。擷取工具列表只用於發現工具，不會執行工具。

mcp = await core.mcp("my-mcp")
available_tools = await mcp.list_tools()
for tool in available_tools:
    print(tool.name, tool.description, tool.parameters)

# 替換為該 MCP 實際提供的工具名稱和參數。
result = await mcp.call_tool("<工具名稱>", {"<參數名稱>": "<參數值>"})

如需讓模型選擇並執行工具，請使用後文的框架組成樣本。模型調用本身不會自動執行 MCP 工具。

使用 Skill

Skill 包含任務說明和可選的輔助檔案或指令碼。載入 Skill 後，通過架構適配器將其接入 Agent。

skill = await core.skills.managed("my-skill", version="1.0.0")
print(skill.name, skill.version)

1.0.0 必須是該 Skill 發行的版本。省略版本時使用平台解析出的預設版本。需要固定應用行為時，建議顯式指定版本。

重要：載入 Skill 不等於執行 Skill。Skill 工具可能執行包內指令碼或命令，僅使用可信來源的 Skill。不需要命令執行能力時，在應用運行環境中設定 ALLOW_EXECUTE_COMMAND=false。Skill 執行所需的程式和依賴仍需包含在應用運行環境中。

使用記憶

MemoryStore 用於儲存和檢索長期記憶。它與模型輸入的對話歷史是不同的能力：檢索到記憶後，需要由業務代碼或架構適配器將其作為參考資訊提供給模型。

from agentcore.memory import MemoryScope

store = core.memory_store("my-memory")
await store.add_memories(
    scope=MemoryScope(user_id="example-user", session_id="example-session"),
    text="使用者喜歡簡短的中文回答。",
)

result = await store.search_memories(
    "使用者有哪些回答偏好？",
    scope=MemoryScope(user_id="example-user"),
    top_k=5,
)
for hit in result.memories:
    print(hit.memory.content.text)

使用憑證和 MCP Header

credential = await core.credentials.get("my-api-key")
api_key = credential.value  # 交給需要該憑證的用戶端，不要輸出。

header_credential = await core.credentials.get("my-mcp-header")
headers = header_credential.as_headers()

mcp = await core.mcp(
    "my-mcp",
    credential_name="my-mcp-header",
    headers={"x-business-id": "my-app"},
)

接入自訂資源

自有模型、MCP 服務和隨應用分發的 Skill 無需註冊為平台資源。以下配置由應用自己提供，樣本中的環境變數不是 SDK 自動讀取的配置項。

import asyncio
import os
from agentcore import AsyncAgentCore

async def main():
    async with AsyncAgentCore() as core:
        model = core.direct_model(
            provider="openai",
            model=os.environ["CUSTOM_MODEL_NAME"],
            base_url=os.environ["CUSTOM_MODEL_BASE_URL"],
            api_key=os.environ["CUSTOM_MODEL_API_KEY"],
        )
        response = await model.invoke([{"role": "user", "content": "你好"}])
        print(response["choices"][0]["message"]["content"])

        mcp = core.direct_mcp(url=os.environ["CUSTOM_MCP_URL"])
        print([tool.name for tool in await mcp.list_tools()])
        skills = await core.skills.local("./skills")
        print([skill.name for skill in skills])

asyncio.run(main())

直連 MCP 需要鑒權時，可通過 headers_provider 提供 Header；stdio 傳輸不使用 HTTP Header。

使用 LangChain 構建 Agent 服務

以下樣本將雲端模型、MCP 和 Skill 接入 LangChain，並通過 AgentCoreServer 提供服務。使用前請準備 my-model-connection、my-mcp、my-skill，並按前文安裝架構依賴。

樣本處理最新一條使用者文本，未實現對話歷史儲存。多輪會話應由應用或架構維護歷史和檢查點。

儲存為 app.py：

import logging
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

from agentcore import AsyncAgentCore
from agentcore.integrations.langchain import AgentCoreConverter, model, skill_tools, tools
from agentcore.server import AgentCoreServer

logging.basicConfig(level=logging.INFO)
core = None
chat = None
agent = None

async def shutdown():
    try:
        if chat is not None:
            await chat.root_async_client.close()
            chat.root_client.close()
    finally:
        if core is not None:
            await core.aclose()

async def startup():
    global core, chat, agent
    core = AsyncAgentCore.auto()
    try:
        mcp = await core.mcp("my-mcp")
        skill = await core.skills.managed("my-skill")
        chat = model("my-model-connection", model_name="qwen3.8-max")
        agent = create_agent(
            model=chat,
            tools=[*tools(await mcp.list_tools()), *skill_tools([skill])],
            system_prompt="使用工具完成請求，不要編造工具執行結果。",
        )
    except Exception:
        logging.exception("Agent 初始化失敗")
        await shutdown()
        raise

server = AgentCoreServer(
    startup=startup,
    shutdown=shutdown,
    readiness=lambda: agent is not None,
)

@server.invoke
async def invoke(request, context):
    if not request.messages:
        raise ValueError("請提供使用者訊息")
    message = request.messages[-1]
    if message.role.value != "user" or not isinstance(message.content, str):
        raise ValueError("本樣本僅支援使用者簡訊")
    events = agent.astream_events(
        {"messages": [HumanMessage(content=message.content)]},
        version="v2",
    )
    async for event in AgentCoreConverter().stream(events):
        yield event

在應用運行環境中使用以下啟動命令：

uvicorn app:server --host 0.0.0.0 --port 9000

樣本按 OpenAI/v1 模型串連編寫，關閉邏輯也對應該樣本的 LangChain OpenAI 用戶端。

調用 Agent 服務

將下面的 https://<agent-endpoint> 替換為應用實際訪問地址，並按應用的訪問要求補充認證資訊。該地址是 Agent 應用地址，不是模型供應商地址。

curl -N 'https://<agent-endpoint>/ag-ui/agent' \
  -H 'Content-Type: application/json' \
  -d '{
    "threadId": "example-thread-1",
    "runId": "example-run-1",
    "messages": [{"id": "message-1", "role": "user", "content": "查看可用 Skill，並按其說明完成一個示範。"}],
    "state": {},


    "tools": [ ],



    "context": [ ],


    "forwardedProps": {}
  }'

curl -N 'https://<agent-endpoint>/openai/v1/chat/completions' \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "agentcore",
    "messages": [{"role": "user", "content": "你好"}],
    "stream": true
  }'

其他框架組成

上述架構入口也提供執行事件轉換器，輸入應使用對應架構的完整事件流。例如 LangChain/LangGraph 使用 astream_events(version="v2")。

CrewAI 完整事件入口需要 CrewAI 1.15.20 及以上版本；其軌跡在執行完成後輸出，不是逐 Token 即時輸出。不同架構的事件輸入和輸出時機不同，不應直接複用另一架構的事件處理方式。

常見問題

相關文檔

• 管理 Workspace

• 管理模型

• 管理 MCP 工具

• 管理 Skill

• Python SDK 安裝包