The AgentCore Python SDK lets you call Workspace resources by name from Python code, including model connections, MCP tools, Skills, memories, and credentials. You can also publish your Agent as an AG-UI or OpenAI Chat Completions service through a framework integration. This topic describes how to install the SDK, call cloud and custom resources, build and deploy an Agent service with LangChain, call the Agent service, and troubleshoot common issues.
An Agent application usually connects to cloud models, external tools, and long-term memory at the same time. Maintaining a separate endpoint, credential, and session state for each capability quickly makes application code hard to maintain. This topic is the Python volume of AgentCore SDK Overview, which covers language selection, capability boundaries, and an overview of the preparations.
This topic covers two resource paths:
-
Cloud resources — Also called platform resources. They are configured in your Workspace, your code calls them by name, and the examples run in the AgentCore managed application runtime environment.
-
Custom resources — Your own models, MCP servers, and Skills that ship with your application. They do not need to be registered as platform resources, and your application provides the connection settings. For details, see "Connect custom resources".
Prerequisites
Runtime environment
| Item | Requirement |
| Python version | The base SDK supports Python 3.10 and later. The LangChain service example in this topic uses Python 3.11 and later. |
| Application runtime environment | The cloud resource examples run in the AgentCore managed application runtime environment. The SDK gets the access configuration from the runtime environment, so you do not need to put platform access keys in your application code. |
| Package and import names | The Python package name is alibabacloud-agentcore-sdk, and the import name in your code is agentcore. |
Platform resources
Before you write code, activate AgentCore and create a Workspace as described in Manage Workspace. All resources in the examples in this topic belong to the Workspace of the Agent. Make sure that you have prepared the required resources and that the Agent has access to them.
This topic uses the following example resource names. Replace them with the names of the resources that you created.
| Resource | Example value | Usage |
| Model connection | my-model-connection |
Calls a model. Enter the name of a model connection that you created as described in Manage Models. |
| Model | qwen3.8-max |
Must already be configured in the selected model connection. The "Build an Agent service with LangChain" example also requires a model that supports tool calls. |
| MCP server | my-mcp |
Calls a tool. Enter the name of an MCP server that you created as described in Manage MCP Servers. |
| Skill | my-skill |
Uses a Skill that is enabled and has a version in the Published state. You create, enable, and publish Skill versions as described in Manage Skill. |
| MemoryStore | my-memory |
Required when you use the memory feature. |
| Credential | my-api-key, my-mcp-header |
Create an API key credential or an MCP header credential as described in Manage credentials, and grant it to your application. |
If you only call models, you do not need to create an MCP server, a Skill, a MemoryStore, or additional credentials.
Install the SDK
Install the base SDK:
pip install alibabacloud-agentcore-sdk==0.1.1
Install the optional dependencies that you need for MCP, HTTP serving, credentials, or framework integrations. The following command uses LangChain as an example. For another framework, replace langchain with the corresponding optional dependency in the following table.
pip install "alibabacloud-agentcore-sdk[mcp,server,credentials,langchain]==0.1.1"
| Optional dependency | Purpose |
mcp |
Connects to MCP servers. |
server |
Uses AgentCoreServer. |
credentials |
Reads managed credentials and binds MCP header credentials. |
langchain, langgraph |
The corresponding framework integrations. |
agentscope |
The AgentScope 2.x integration. Requires Python 3.11 and later. |
google-adk, pydantic-ai, crewai |
The corresponding framework integrations. The frameworks themselves can have additional Python version requirements. The complete event entry point of CrewAI also requires CrewAI 1.15.20 and later. |
The examples in this topic use Python SDK 0.1.1. For later versions, see the alibabacloud-agentcore-sdk PyPI package page.
Create and manage Core
The cloud resource snippets in this topic run within the lifecycle of an existing core. Create Core in one of the following ways:
-
Cloud resource examples use
async with AsyncAgentCore.auto() as core. For the complete example, see "Call cloud models". The SDK gets the access configuration from the application runtime environment, so you do not need to put platform access keys in your application code. -
Custom resource examples construct Core directly with
Reuse one Core within the application lifecycle and close it when the application exits. (Recommended) Use an async context manager (async with AsyncAgentCore() as core, and your application provides the connection settings. For details, see "Connect custom resources".async with AsyncAgentCore.auto() as core), or callawait core.aclose(). Do not close a shared Core at the end of each HTTP request while other requests are still using it.
Call cloud models
Specify the model connection name and the model name in your code. The SDK looks up the model connection in the Workspace that the current Agent belongs to, so you do not need to construct the model endpoint yourself.
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": "Describe Hangzhou in one sentence."}],
temperature=0.2,
)
print(response["choices"][0]["message"]["content"])
asyncio.run(main())
| Parameter | Description |
my-model-connection |
The name of the model connection. This is not the model name or the model connection ID. |
model |
The name of a model that is already configured in that connection. |
temperature |
A generation parameter for this request. Whether the model supports this parameter and its valid range depend on the model. |
-
OpenAI/v1 connections can use
responses()andresponses_stream(). The model must support the Responses API, and the SDK does not fall back to Chat Completions when a call fails. -
The managed model client does not provide embedding calls. The
embedding()method of a custom model client works only with services and models that support vector generation.
Streaming output
Within the Core lifecycle described in "Create and manage Core", replace the regular call with the following code. A regular call and a streaming call are two separate requests, so choose one based on your needs.
async for chunk in model.stream([{"role": "user", "content": "Describe Hangzhou in one sentence."}]):
for choice in chunk.get("choices", []):
print(choice.get("delta", {}).get("content") or "", end="", flush=True)
The text parsing example above applies to the OpenAI/v1 protocol. The response data structure is determined by the protocol of the model connection. Anthropic models return the data structure of their own protocol, so do not parse them with choices.
Use MCP tools
The following snippet runs within the lifecycle of the core that you created in "Create and manage Core". Listing tools only discovers them; it does not run any tool.
mcp = await core.mcp("my-mcp")
available_tools = await mcp.list_tools()
for tool in available_tools:
print(tool.name, tool.description, tool.parameters)
# Replace these with the tool name and parameters that the MCP server actually provides.
result = await mcp.call_tool("<tool-name>", {"<parameter-name>": "<parameter-value>"})
To let the model select and run tools, pass the tools to your Agent framework as shown in the "Build an Agent service with LangChain" example. A model call by itself does not run MCP tools. If an MCP server requires authentication, bind a credential to the client as described in "Bind a credential to an MCP server".
Use Skills
A Skill contains task instructions and optional supporting files or scripts. After you load a Skill, connect it to your Agent through a framework integration. For the complete integration, see the "Build an Agent service with LangChain" example.
skill = await core.skills.managed("my-skill", version="1.0.0")
print(skill.name, skill.version)
1.0.0 must be a version of the Skill that is in the Published state. If you omit the version, the platform resolves the default version. Specify a version explicitly when you need to keep the behavior of your application fixed.
Loading a Skill is not the same as running it. Skill tools can run the scripts or commands inside the package, so use only Skills from trusted sources. If you do not need command execution, set ALLOW_EXECUTE_COMMAND=false in the application runtime environment. The programs and dependencies that a Skill needs must still be included in the application runtime environment.
Use memory
A MemoryStore saves and retrieves long-term memory. It is a different capability from the conversation history that you send to a model: after you retrieve memories, your application code or a framework integration must pass them to the model as reference information.
Write and retrieve memories
The following snippet runs within the lifecycle of the core that you created in "Create and manage Core". The example writes data, so use a MemoryStore that is suitable for testing.
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="The user prefers short answers.",
)
result = await store.search_memories(
"What response preferences does the user have?",
scope=MemoryScope(user_id="example-user"),
top_k=5,
)
for hit in result.memories:
print(hit.memory.content.text)
| Parameter | Description |
user_id |
The logical identifier of a business user. Use it to organize memories by user. |
agent_id |
The logical identifier of an Agent. Define it based on your business needs. You do not need to provide it in every call. |
session_id |
The session identifier. When you write a memory, it marks the source of the memory. When you search, passing it limits the results to that session. |
top_k |
The maximum number of results that a search returns. |
Usage notes:
-
The Workspace that the MemoryStore belongs to comes from the resource context of the Core, so you usually do not need to pass it again in each memory call.
-
When you write, you can omit all three scope fields, and the fields that you do not pass fall into the server-side default scope. When you search, a field that you do not pass does not limit that dimension. In a multi-user application, pass the scopes that you need explicitly to avoid retrieving memories that should not be shared.
-
To use user preferences across sessions, write memories by user and do not limit
session_idwhen you search. -
Scope fields must come from a trusted business context. Do not let a model generate or change them freely. They provide logical partitioning and are not a replacement for access credentials or permission checks.
-
A newly written memory can take some time before it becomes searchable. Do not write the same content repeatedly just because the first search returns nothing.
-
To query session messages, use
list_memory_session_messages(). This method requires a session ID and at least one of the user ID and the Agent ID.
Integrate memory into the framework execution flow
A framework integration can retrieve memories before the model runs and write memories back after the run, based on its configuration. Connecting only a model or tool integration does not enable memory automatically. For the memory entry point of each framework, see the framework table in "Other framework integrations".
Each framework has a different lifecycle, so connect memory writes to the corresponding middleware, node, or session persistence interface. Do not submit the entire history again after every model call.
Use credentials and MCP headers
Retrieve managed credentials
After you create a credential on the platform and complete authorization, you can read it by name. The following snippet runs within the lifecycle of the core that you created in "Create and manage Core".
credential = await core.credentials.get("my-api-key")
api_key = credential.value # Pass this to the client that needs the credential. Do not print it.
header_credential = await core.credentials.get("my-mcp-header")
headers = header_credential.as_headers()
Use value for an API key credential and as_headers() for an MCP header credential. Do not write credential values into source code, logs, or external responses.
Bind a credential to an MCP server
mcp = await core.mcp(
"my-mcp",
credential_name="my-mcp-header",
headers={"x-business-id": "my-app"},
)
Both credential_name and headers are optional. The MCP header credential that you bind must be allowed to apply to that MCP server.
-
headersis a fixed client-level setting that applies to the MCP handshake, tool discovery, and tool calls. It is not a per-request user identity parameter, so do not change headers on a shared client to switch users. -
Header names are case-insensitive.
-
A conflict between platform headers, credential headers, and custom headers raises an error instead of overwriting a value.
-
You cannot overwrite platform authentication fields such as
Authorization, and you cannot manually set protocol fields such asHostandMcp-Session-Id. -
The SDK does not automatically copy all inbound headers.
Connect custom resources
Your own models, MCP servers, and Skills that ship with your application do not need to be registered as platform resources. Your application provides the following settings, and the environment variables in the example are not configuration items that the SDK reads automatically.
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": "Hello"}])
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())
| Setting | Description |
CUSTOM_MODEL_NAME |
The name of a model that your own service supports. |
CUSTOM_MODEL_BASE_URL |
The base URL of the model API. An OpenAI-compatible service usually includes /v1. Follow the documentation of your service. |
CUSTOM_MODEL_API_KEY |
The API key of the model service. Provide it through the runtime environment or a secrets manager. |
CUSTOM_MCP_URL |
The full access URL of the MCP server. The default transport is Streamable HTTP. For an SSE service, set transport="sse" explicitly. |
./skills |
The directory of Skills that ship with your application. Each Skill must contain a SKILL.md file and the files that it references. |
When a directly connected MCP server requires authentication, provide headers through headers_provider. The stdio transport does not use HTTP headers.
Build an Agent service with LangChain
AgentCoreServer is not a prerequisite for calling cloud resources. You can use the model, MCP, Skill, memory, and credential interfaces in an existing web service or Agent executor, and AgentCoreServer only reuses its protocol encapsulation.
The following example connects a cloud model, an MCP server, and a Skill to LangChain and serves them through AgentCoreServer. Before you start, prepare my-model-connection, my-mcp, and my-skill, and install the framework dependencies described in "Install the SDK".
The example handles only the latest user text message and does not store conversation history. For multi-turn conversations, your application or framework must maintain the history and the checkpoints.
Save the following code as 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="Use the tools to complete the request. Do not fabricate tool results.",
)
except Exception:
logging.exception("Agent initialization failed")
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("Provide a user message.")
message = request.messages[-1]
if message.role.value != "user" or not isinstance(message.content, str):
raise ValueError("This example supports only user text messages.")
events = agent.astream_events(
{"messages": [HumanMessage(content=message.content)]},
version="v2",
)
async for event in AgentCoreConverter().stream(events):
yield event
Run the following start command in the application runtime environment:
uvicorn app:server --host 0.0.0.0 --port 9000
The example is written for an OpenAI/v1 model connection, and the shutdown logic matches the LangChain OpenAI client in this example.
Deploy the Agent service
Package your application code and dependencies and deploy them to AgentCore. The application runtime environment must include the SDK and framework dependencies described in "Install the SDK". If you use a container image, install the dependencies when you build the image.
| Setting | Value in this topic |
| Listen address | 0.0.0.0 |
| Service port | 9000 |
| Python start command | uvicorn app:server --host 0.0.0.0 --port 9000 |
| Liveness check | GET /healthz |
| Readiness check | GET /readyz |
The port that you configure on the platform must match the port that your application listens on. The SDK does not upload code, build images, or deploy applications.
Call the Agent service
Replace https://<agent-endpoint> in the following examples with the actual access address of your application, and add the authentication information that your application requires. This address is the address of the Agent application, not the address of a model provider.
Choose a protocol
Choose a service protocol based on your client type and on how you want to display the run:
| Capability | AG-UI | OpenAI Chat Completions |
| Text output | Supported. | Supported. |
| Multiple message boundaries | Keeps the start, content, and end events of a text message. | Aggregates the text of one completion into a single assistant message. |
| Tool calls | Emits tool call events and keeps the call ID. | Emits tool_calls. |
| Tool results | Emits separate tool result events. | Does not provide separate tool result response events. |
| Use cases | Shows the multi-step execution process of an Agent, including tool calls and results. | Integrates clients that use the Chat Completions format. |
The SSE stream of both protocols sends a heartbeat comment after 15 seconds of idle output. A heartbeat is not model output and does not mean that the task has finished.
The framework service example already runs tools on the server side, so the client must not run the received tool call trace again. When you need the complete execution process, pass the complete event stream of the framework to AgentCoreConverter instead of extracting only the text, and create a separate converter for each request. The converter keeps structured message boundaries and tool correlations, and does not guess the message type from the message body.
AG-UI
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": "List the available Skills and complete a demonstration as instructed."}],
"state": {},
"tools": [],
"context": [],
"forwardedProps": {}
}'
Use a new runId for each run. You can reuse the same threadId for one conversation, but that does not make the preceding example save history automatically. The "tools": [] field in the request does not disable the tools that are configured in the Agent code.
A successful run starts with RUN_STARTED and ends with RUN_FINISHED. When a tool call occurs, the service emits tool call and tool result events and correlates them through the same toolCallId. Whether a tool is called depends on the model and the task. A failed run emits RUN_ERROR.
OpenAI Chat Completions
curl -N 'https://<agent-endpoint>/openai/v1/chat/completions' \
-H 'Content-Type: application/json' \
-d '{
"model": "agentcore",
"messages": [{"role": "user", "content": "Hello"}],
"stream": true
}'
A streaming response ends with [DONE]. Set stream to false to get a single JSON response.
model is a required field in the OpenAI request format. This example always uses the model connection that the application code specifies and does not switch the downstream model based on this field. To query the protocol model list of the application, use GET /openai/v1/models. This list is not the model catalog of the Workspace.
Other framework integrations
The following table is the complete reference for all supported frameworks, including the LangChain framework used in "Build an Agent service with LangChain". It lists the optional dependency, the framework integration entry point, and the memory integration entry point of each framework:
| Framework | Optional dependency | Framework integration entry point | Memory integration entry point |
| LangChain | langchain |
agentcore.integrations.langchain |
agentcore.integrations.memory.langchain |
| LangGraph | langgraph |
agentcore.integrations.langgraph |
agentcore.integrations.memory.langgraph |
| AgentScope 2.x | agentscope |
agentcore.integrations.agentscope |
agentcore.integrations.memory.agentscope |
| Google ADK | google-adk |
agentcore.integrations.google_adk |
agentcore.integrations.memory.google_adk |
| PydanticAI | pydantic-ai |
agentcore.integrations.pydantic_ai |
agentcore.integrations.memory.pydantic_ai |
| CrewAI | crewai |
agentcore.integrations.crewai |
agentcore.integrations.memory.crewai |
These framework entry points also provide execution event converters, and the input must be the complete event stream of the corresponding framework. LangChain and LangGraph use astream_events(version="v2").
The complete event trace of CrewAI is emitted after the run finishes instead of token by token in real time. Different frameworks have different event inputs and output timing, so do not reuse the event handling approach of another framework directly.
FAQ
What do I do if a model connection, MCP server, or Skill cannot be found?
Check the Workspace that the application belongs to, the resource names, and the access permissions. The model connection name and the model name are two different parameters. For a Skill, also check whether it is enabled and whether the selected version is published. Do not substitute the resource ID of another Workspace for a name.
Why does the model not call tools after I load MCP tools and Skills?
list_tools() and Skill loading only prepare the resources. You must also pass the tools to the Agent framework and select a model that supports tool calls. Even when tools are configured, a simple question might not require a tool call.
Why is a newly written memory not returned by the first search?
Memory processing and the searchable state can be delayed. First check whether the write succeeded and whether the read and write scopes match, and then query again later. Do not replay the write operation.
Does the same session ID automatically restore history?
No. AgentCoreServer handles protocol access and does not persist conversation history for your application. Your application or framework manages the session state, and long-term memory is not a replacement for complete conversation history or framework checkpoints.
What do I do if cross-origin requests from a browser fail?
The SDK does not open cross-origin access by default. Your application or the ingress gateway must configure CORS based on the actual frontend origins. Do not allow all origins as the default configuration for interfaces that use sensitive credentials.
How do I log troubleshooting information?
Use logging.basicConfig(level=logging.INFO) to enable logging. Keep error stacks, request IDs, and failed operations, and avoid printing access keys, complete headers, or sensitive user content.
References
-
AgentCore SDK Overview: an overview of SDK capabilities, development language selection, and service protocol comparison.
-
Manage Workspace: create and switch Workspaces. All resources in the examples in this topic belong to a Workspace.
-
Manage Models: add a model connection and configure models in that connection.
-
Manage MCP Servers: create and manage MCP servers for a Workspace.
-
Manage Skill: create a Skill, enable it, and publish a version.
-
Manage credentials: create an API key credential or an MCP header credential and grant it to your application.
-
alibabacloud-agentcore-sdk PyPI package page: check the latest version of the Python SDK.
-
AgentCore product page: learn about the positioning and core features of AgentCore.