Build and Deploy MCP Server with Amazon Cognito OAuth on Amazon Bedrock AgentCore

import os
import boto3
from mcp.server import MCPServer

mcp = MCPServer(
"aws-resource-mcp-server",
version="1.0.0",
)

AWS_REGION = os.environ.get(
"AWS_REGION",
"eu-west-2",
)

@mcp.tool()
def get_aws_identity() -> dict:
"""
Get the AWS identity used by the MCP Server.
"""

sts = boto3.client(
"sts",
region_name=AWS_REGION,
)

response = sts.get_caller_identity()

return {
"account": response["Account"],
"arn": response["Arn"],
"user_id": response["UserId"],
}

@mcp.tool()
def list_s3_buckets() -> list[dict]:
"""
List Amazon S3 buckets in the AWS account.
"""

s3 = boto3.client("s3")

response = s3.list_buckets()

buckets = []

for bucket in response.get(
"Buckets",
[],
):
buckets.append(
{
"name": bucket["Name"],
"creation_date": (
bucket[
"CreationDate"
].isoformat()
),
}
)

return buckets

@mcp.tool()
def list_ec2_instances(
region: str = AWS_REGION,
) -> list[dict]:
"""
List Amazon EC2 instances
in the specified AWS Region.
"""

ec2 = boto3.client(
"ec2",
region_name=region,
)

paginator = ec2.get_paginator(
"describe_instances"
)

instances = []

for page in paginator.paginate():
for reservation in page.get(
"Reservations",
[],
):
for instance in reservation.get(
"Instances",
[],
):
name = ""

for tag in instance.get(
"Tags",
[],
):
if tag.get("Key") == "Name":
name = tag.get(
"Value",
"",
)
break

instances.append(
{
"instance_id": (
instance["InstanceId"]
),
"name": name,
"instance_type": (
instance["InstanceType"]
),
"state": (
instance[
"State"
]["Name"]
),
"private_ip": (
instance.get(
"PrivateIpAddress"
)
),
"public_ip": (
instance.get(
"PublicIpAddress"
)
),
"region": region,
}
)

return instances

if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
stateless_http=True,
)