bedrock agentcore iam runtime MCP server

在Amazon Bedrock AgentCore上使用IAM认证的MCP服务器

本指南展示了如何使用IAM认证（而非OAuth/Cognito令牌）在Amazon Bedrock AgentCore Runtime上部署一个模型上下文协议（MCP）服务器。

先决条件

• Python 3.10或更高版本
• 具有适当权限的AWS账户
• 使用管理员凭据配置的AWS CLI

项目结构

bedrock_agentcore_iam_runtime_mcp_server/
├── server/
│   ├── my_iam_mcp_server.py    # MCP server implementation
│   └── requirements.txt        # Server dependencies
├── client/
│   ├── test_mcp_client.py     # Test client
│   └── requirements.txt        # Client dependencies
├── iam/
│   └── mcp-access-policy.json # IAM policy for MCP access
└── README.md                  # This file

注如果你正在克隆这个仓库，由于文件已经创建好了，你可以直接跳到第3步（以及之后的几步）。

步骤1：创建您的MCP服务器

创建 server/my_iam_mcp_server.py：

from mcp.server.fastmcp import FastMCP
import boto3

mcp = FastMCP("my_iam_mcp_server", host="0.0.0.0", stateless_http=True)

@mcp.tool()
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

@mcp.tool()
def multiply_numbers(a: int, b: int) -> int:
    """Multiply two numbers together"""
    return a * b

@mcp.tool()
def greet_user(name: str) -> str:
    """Greet a user by name"""
    return f"Hello, {name}! Nice to meet you."

@mcp.tool()
def get_aws_region() -> str:
    """Get the current AWS region using boto3"""
    session = boto3.Session()
    return session.region_name

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

步骤2：创建需求文件

创建 server/requirements.txt：

mcp
boto3
bedrock-agentcore
bedrock-agentcore-starter-toolkit

创建 client/requirements.txt:

mcp
boto3
run-mcp-servers-with-aws-lambda

第三步：配置并部署

导航至服务器目录并配置您的MCP服务器：

cd server
agentcore configure -e my_iam_mcp_server.py --protocol MCP

在配置过程中：

• 执行角色按回车键自动创建
• ECR 仓库按回车键自动创建
• 依赖文件按回车键使用检测结果 requirements.txt
• 授权选择 no 用于 OAuth（默认使用 IAM）

部署到AWS：

agentcore launch

成功部署后，您将收到一个类似如下的代理ARN（Amazon Resource Name）：

arn:aws:bedrock-agentcore:::runtime/my_iam_mcp_server-

保存这个ARN——下一步需要用到它。

步骤4：使用您当前的凭据进行测试

导航到客户端目录并安装依赖项：

cd ../client
pip install -r requirements.txt

创造 client/test_mcp_client.py：

import boto3
import asyncio
from mcp import ClientSession
from mcp_lambda.client.streamable_http_sigv4 import streamablehttp_client_with_sigv4

def generate_mcp_url(agent_runtime_arn: str, region: str = "") -> str:
    encoded_arn = agent_runtime_arn.replace(':', '%3A').replace('/', '%2F')
    return f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"

async def test_mcp_server():
    # Replace with your actual Agent ARN
    agent_arn = "arn:aws:bedrock-agentcore:::runtime/my_iam_mcp_server-"
    
    mcp_url = generate_mcp_url(agent_arn, region="")
    print(f"Connecting to: {mcp_url}")

    session = boto3.Session()
    credentials = session.get_credentials()
    
    try:
        async with streamablehttp_client_with_sigv4(
            url=mcp_url,
            service="bedrock-agentcore",
            region="",
            credentials=credentials,
            timeout=120,
            terminate_on_close=False
        ) as (read_stream, write_stream, _):
            async with ClientSession(read_stream, write_stream) as mcp_session:
                print("Initializing MCP session...")
                await mcp_session.initialize()
                print("MCP session initialized successfully")
                
                # List available tools
                print("\n=== Available Tools ===")
                tool_result = await mcp_session.list_tools()
                for tool in tool_result.tools:
                    print(f"  - {tool.name}: {tool.description}")
                
                # Test the tools
                print("\n=== Testing add_numbers tool ===")
                result = await mcp_session.call_tool("add_numbers", {"a": 5, "b": 3})
                print(f"add_numbers(5, 3) = {result.content}")
                
                print("\n=== Testing multiply_numbers tool ===")
                result = await mcp_session.call_tool("multiply_numbers", {"a": 4, "b": 7})
                print(f"multiply_numbers(4, 7) = {result.content}")
                
                print("\n=== Testing greet_user tool ===")
                result = await mcp_session.call_tool("greet_user", {"name": "Alice"})
                print(f"greet_user('Alice') = {result.content}")
                
                # Test the boto3 tool
                print("\n=== Testing get_aws_region tool (uses boto3) ===")
                result = await mcp_session.call_tool("get_aws_region", {})
                print(f"get_aws_region() = {result.content}")
                
    except Exception as e:
        print(f"Error connecting to MCP server {e}")
        raise

if __name__ == "__main__":
    asyncio.run(test_mcp_server())

在运行之前，请更新以下占位符： test_mcp_client.py：

• 替换 ` 与您的AWS区域（例如。， us-east-1`）
• 替换 `` 使用您的AWS账户ID
• 替换 `` 使用来自您代理ARN的随机ID

使用您当前的凭据进行测试：

python3 test_mcp_client.py

步骤5：为测试创建独立的IAM用户

创建IAM用户

# Create the IAM user
aws iam create-user --user-name mcp-test-user

# Create access keys for the user
aws iam create-access-key --user-name mcp-test-user

保存 AccessKeyId 并且 SecretAccessKey 从输出中。

创建IAM策略

创建 iam/mcp-access-policy.json：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "bedrock-agentcore:InvokeAgentRuntime"
            ],
            "Resource": "arn:aws:bedrock-agentcore:::runtime/my_iam_mcp_server-*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "bedrock-agentcore:InvokeAgentRuntime"
            ],
            "Resource": "arn:aws:bedrock-agentcore:::runtime/*"
        }
    ]
}

在创建策略之前，请更新以下占位符： mcp-access-policy.json：

• 替换 ` 与您的AWS区域（例如。， us-east-1`）
• 替换 `` 使用您的AWS账户ID

创建并附加策略：

# Create the policy
aws iam create-policy \
    --policy-name MCPServerAccessPolicy \
    --policy-document file://iam/mcp-access-policy.json

# Attach the policy to the user (replace  with your AWS account ID)
aws iam attach-user-policy \
    --user-name mcp-test-user \
    --policy-arn arn:aws:iam:::policy/MCPServerAccessPolicy

在运行attach命令之前，请替换 `` 使用您的AWS账户ID。

使用新IAM用户进行测试

打开一个新的终端，以避免使用您的管理员凭据，并使用新用户的凭据设置您的环境变量：

export AWS_ACCESS_KEY_ID=""
export AWS_SECRET_ACCESS_KEY=""
export AWS_DEFAULT_REGION=""

# Run the test
python3 test_mcp_client.py

在跑步之前，请更换以下物品：

• `` 使用create-access-key命令生成的AccessKeyId
• `` 使用从create-access-key命令获取的SecretAccessKey
• ` 与您的AWS区域（例如。， us-east-1`)

预期输出

Connecting to: https://bedrock-agentcore..amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3A%3A%3Aruntime%2Fmy_iam_mcp_server-/invocations?qualifier=DEFAULT

Initializing MCP session...
MCP session initialized successfully

=== Available Tools ===
  - add_numbers: Add two numbers together
  - multiply_numbers: Multiply two numbers together
  - greet_user: Greet a user by name
  - get_aws_region: Get the current AWS region using boto3

=== Testing add_numbers tool ===
add_numbers(5, 3) = [TextContent(type='text', text='8', annotations=None, meta=None)]

=== Testing multiply_numbers tool ===
multiply_numbers(4, 7) = [TextContent(type='text', text='28', annotations=None, meta=None)]

=== Testing greet_user tool ===
greet_user('Alice') = [TextContent(type='text', text='Hello, Alice! Nice to meet you.', annotations=None, meta=None)]

=== Testing get_aws_region tool (uses boto3) ===
get_aws_region() = [TextContent(type='text', text='us-east-1', annotations=None, meta=None)]

主要优势

• 无需设置OAuth/Cognito - 使用标准的AWS IAM认证
• SigV4 签名 - 使用boto3凭据自动对AWS请求进行签名
• 分离的依赖项 - 服务器和客户端的依赖关系是隔离的
• boto3 支持 - 在MCP服务器工具中演示使用AWS SDK

参考文献

• AWS Bedrock AgentCore MCP 文档
• AWS Bedrock AgentCore MCP 服务器教程