MCP服务器部署工具

该工具用于在Amazon Bedrock AgentCore Runtime上部署MCP服务器,支持IAM认证,提供数学计算、用户问候和AWS区域获取等功能。

  • by des1-gner
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 des1-gner/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:<aws-region>:<account-id>:runtime/my_iam_mcp_server-<random-id>

保存这个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 = "<aws-region>") -> 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:<aws-region>:<account-id>:runtime/my_iam_mcp_server-<random-id>"
    
    mcp_url = generate_mcp_url(agent_arn, region="<aws-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="<aws-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-region> 与您的AWS区域(例如。, us-east-1)
  • 替换 <account-id> 使用您的AWS账户ID
  • 替换 <random-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:<aws-region>:<account-id>:runtime/my_iam_mcp_server-*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "bedrock-agentcore:InvokeAgentRuntime"
            ],
            "Resource": "arn:aws:bedrock-agentcore:<aws-region>:<account-id>:runtime/*"
        }
    ]
}

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

  • 替换 <aws-region> 与您的AWS区域(例如。, us-east-1)
  • 替换 <account-id> 使用您的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 <account-id> with your AWS account ID)
aws iam attach-user-policy \
    --user-name mcp-test-user \
    --policy-arn arn:aws:iam::<account-id>:policy/MCPServerAccessPolicy

在运行attach命令之前,请替换 <account-id> 使用您的AWS账户ID。

使用新IAM用户进行测试

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

export AWS_ACCESS_KEY_ID="<access-key-id>"
export AWS_SECRET_ACCESS_KEY="<secret-access-key>"
export AWS_DEFAULT_REGION="<aws-region>"

# Run the test
python3 test_mcp_client.py

在跑步之前,请更换以下物品:

  • <access-key-id> 使用create-access-key命令生成的AccessKeyId
  • <secret-access-key> 使用从create-access-key命令获取的SecretAccessKey
  • <aws-region> 与您的AWS区域(例如。, us-east-1)

预期输出

Connecting to: https://bedrock-agentcore.<aws-region>.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3A<aws-region>%3A<account-id>%3Aruntime%2Fmy_iam_mcp_server-<random-id>/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

参考文献