智能体构建和治理平台：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 安装包