Build a Secure MCP Server on AWS with Amazon Bedrock AgentCore

An MCP server becomes much more useful when it can reach real systems. It also becomes much more dangerous if it receives broad credentials or exposes mutation tools by accident.

In this tutorial, we will build a deliberately small remote MCP server that can:

• list text documents beneath one prefix in one Amazon S3 bucket;
• read a selected UTF-8 document;
• reject path escapes, binary files, and oversized reads;
• authenticate remote callers with Amazon Cognito; and
• run as a stateless Streamable HTTP server on Amazon Bedrock AgentCore Runtime.

The result has two MCP tools—list_documents and read_document—and no upload, overwrite, or delete capability.

Validation note: I locally tested the sample on August 18, 2026 with Python 3.12.3, mcp 1.29.0, and boto3 1.43.73. All 17 unit tests passed, followed by a real Streamable HTTP initialization, tool-discovery, and tool-call flow against a stubbed S3 backend. I did not create or deploy AWS resources while preparing this article; the AWS commands below were checked against the current official documentation.

Architecture

The request flow is:

1. Amazon Cognito authenticates the user and issues a signed access token.
2. The MCP client calls the AgentCore Runtime endpoint over HTTPS using MCP Streamable HTTP and the bearer token.
3. AgentCore validates the token against Cognito's OIDC discovery document and routes the MCP request to FastMCP at 0.0.0.0:8000/mcp.
4. FastMCP invokes a read-only Python tool. Boto3 automatically uses the runtime's temporary IAM credentials.
5. The IAM role permits only s3:ListBucket and s3:GetObject beneath the configured prefix. Runtime telemetry goes to Amazon CloudWatch.

Streamable HTTP is MCP's standard remote transport. Each JSON-RPC message is sent with HTTP POST, and a server can return either JSON or an SSE stream. The client SDK handles the required media types and MCP headers for us. See the MCP transport specification for the wire-level contract.

AgentCore Runtime is useful here because it already understands MCP runtimes, sessions, scaling, security isolation, and observability. This design therefore does not need a custom API Gateway-to-Lambda streaming adapter. AWS documents the runtime contract in Deploy MCP servers in AgentCore Runtime.

Prerequisites

You need:

• an AWS account and an AWS CLI v2 profile with permission to create Cognito, IAM, S3, CloudFormation, and AgentCore resources;
• Python 3.12;
• Node.js 20 or newer and npm;
• jq;
• the AWS CDK; and
• the AgentCore CLI.

Install the two CLIs and verify your AWS identity:

npm install --global aws-cdk @aws/agentcore
aws sts get-caller-identity
agentcore --help

Use one region consistently. This tutorial defaults to us-west-2:

export AWS_REGION="us-west-2"
export AWS_ACCOUNT_ID="$(aws sts get-caller-identity --query Account --output text)"

Cost and cleanup warning: The commands in the deployment sections create billable AWS resources. Run the cleanup section when you finish. Do not paste real passwords, tokens, account IDs, or generated resource names into a public article, screenshot, repository, or issue.

1. Create the tutorial S3 bucket

Create a globally unique bucket name from your account and region:

export S3_BUCKET="s3-document-mcp-${AWS_ACCOUNT_ID}-${AWS_REGION}"
export S3_PREFIX="documents/"

if [ "$AWS_REGION" = "us-east-1" ]; then
  aws s3api create-bucket \
    --bucket "$S3_BUCKET" \
    --region "$AWS_REGION"
else
  aws s3api create-bucket \
    --bucket "$S3_BUCKET" \
    --region "$AWS_REGION" \
    --create-bucket-configuration "LocationConstraint=$AWS_REGION"
fi

aws s3api put-public-access-block \
  --bucket "$S3_BUCKET" \
  --public-access-block-configuration \
  'BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true'

Upload one sample document:

mkdir -p tutorial-docs
printf '# Welcome\n\nThis document came from the protected S3 prefix.\n' > tutorial-docs/welcome.md
aws s3 cp tutorial-docs/welcome.md "s3://${S3_BUCKET}/${S3_PREFIX}welcome.md"

The bucket remains private. The MCP server will later receive only prefix-scoped read permission.

2. Build the FastMCP server

Create a source directory and virtual environment:

mkdir -p s3-document-mcp-source
cd s3-document-mcp-source
python3.12 -m venv .venv
source .venv/bin/activate

Create requirements.txt:

mcp>=1.29,<2
boto3>=1.43,<2

The <2 constraint is intentional. AWS's current FastMCP example imports mcp.server.fastmcp; the MCP 2.0 Python package reorganized that API. Keep the compatible 1.x line until your implementation and the AgentCore documentation move to the newer API together.

Install the dependencies:

python -m pip install --upgrade pip
python -m pip install -r requirements.txt

Create server.py:

"""Read-only S3 document tools exposed over MCP Streamable HTTP."""

from __future__ import annotations

import os
from dataclasses import dataclass
from datetime import datetime
from pathlib import PurePosixPath
from typing import Any, Protocol

import boto3
from botocore.exceptions import ClientError
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from mcp.types import ToolAnnotations


DEFAULT_PREFIX = "documents/"
DEFAULT_MAX_DOCUMENT_BYTES = 65_536
MAX_LIST_LIMIT = 50
ALLOWED_EXTENSIONS = {".csv", ".json", ".md", ".txt", ".yaml", ".yml"}


class S3Client(Protocol):
    def list_objects_v2(self, **kwargs: Any) -> dict[str, Any]: ...

    def get_object(self, **kwargs: Any) -> dict[str, Any]: ...


@dataclass(frozen=True)
class Settings:
    bucket: str
    prefix: str = DEFAULT_PREFIX
    region: str = "us-west-2"
    max_document_bytes: int = DEFAULT_MAX_DOCUMENT_BYTES

    @classmethod
    def from_env(cls) -> "Settings":
        bucket = os.getenv("S3_BUCKET", "").strip()
        if not bucket:
            raise ToolError("S3_BUCKET is required before document tools can run.")

        raw_limit = os.getenv("MAX_DOCUMENT_BYTES", str(DEFAULT_MAX_DOCUMENT_BYTES))
        try:
            max_document_bytes = int(raw_limit)
        except ValueError as exc:
            raise ToolError("MAX_DOCUMENT_BYTES must be a positive integer.") from exc
        if max_document_bytes < 1:
            raise ToolError("MAX_DOCUMENT_BYTES must be a positive integer.")

        prefix = normalize_relative_path(
            os.getenv("S3_PREFIX", DEFAULT_PREFIX),
            allow_empty=True,
            field="S3_PREFIX",
        )
        if prefix:
            prefix = f"{prefix}/"

        return cls(
            bucket=bucket,
            prefix=prefix,
            region=os.getenv("AWS_REGION", "us-west-2").strip() or "us-west-2",
            max_document_bytes=max_document_bytes,
        )


def normalize_relative_path(value: str, *, allow_empty: bool, field: str) -> str:
    candidate = value.strip().replace("\\", "/")
    if candidate.startswith("/"):
        raise ToolError(f"{field} must be relative to the configured S3 prefix.")

    parts = [part for part in candidate.split("/") if part]
    if any(part in {".", ".."} for part in parts):
        raise ToolError(f"{field} cannot contain '.' or '..' path segments.")
    if not parts and not allow_empty:
        raise ToolError(f"{field} cannot be empty.")
    return "/".join(parts)


def resolve_key(settings: Settings, relative_key: str) -> str:
    relative = normalize_relative_path(relative_key, allow_empty=False, field="key")
    return f"{settings.prefix}{relative}"


def build_s3_client(settings: Settings) -> S3Client:
    return boto3.client("s3", region_name=settings.region)


def _relative_key(settings: Settings, full_key: str) -> str | None:
    if settings.prefix and not full_key.startswith(settings.prefix):
        return None
    relative = full_key[len(settings.prefix) :] if settings.prefix else full_key
    return relative or None


def _isoformat(value: Any) -> str:
    if isinstance(value, datetime):
        return value.isoformat()
    return str(value)


def list_documents_impl(
    prefix: str,
    limit: int,
    *,
    settings: Settings,
    client: S3Client,
) -> list[dict[str, Any]]:
    if not 1 <= limit <= MAX_LIST_LIMIT:
        raise ToolError(f"limit must be between 1 and {MAX_LIST_LIMIT}.")

    relative_prefix = normalize_relative_path(prefix, allow_empty=True, field="prefix")
    full_prefix = settings.prefix + relative_prefix
    results: list[dict[str, Any]] = []
    continuation_token: str | None = None

    try:
        while len(results) < limit:
            request: dict[str, Any] = {
                "Bucket": settings.bucket,
                "Prefix": full_prefix,
                "MaxKeys": limit - len(results),
            }
            if continuation_token:
                request["ContinuationToken"] = continuation_token

            response = client.list_objects_v2(**request)
            for item in response.get("Contents", []):
                full_key = str(item.get("Key", ""))
                relative_key = _relative_key(settings, full_key)
                if not relative_key or relative_key.endswith("/"):
                    continue
                results.append(
                    {
                        "key": relative_key,
                        "size": int(item.get("Size", 0)),
                        "last_modified": _isoformat(item.get("LastModified", "")),
                    }
                )
                if len(results) == limit:
                    break

            if not response.get("IsTruncated") or len(results) == limit:
                break
            continuation_token = response.get("NextContinuationToken")
            if not continuation_token:
                break
    except ClientError as exc:
        raise _safe_s3_error(exc) from exc

    return results


def _decode_utf8_prefix(payload: bytes, max_bytes: int) -> str:
    limited = payload[:max_bytes]
    for trim in range(0, min(3, len(limited)) + 1):
        candidate = limited if trim == 0 else limited[:-trim]
        try:
            return candidate.decode("utf-8")
        except UnicodeDecodeError as exc:
            near_boundary = exc.start >= max(0, len(candidate) - 3)
            if not near_boundary or trim == min(3, len(limited)):
                raise ToolError("The requested document is not valid UTF-8 text.") from exc
    raise ToolError("The requested document is not valid UTF-8 text.")


def read_document_impl(
    key: str,
    *,
    settings: Settings,
    client: S3Client,
) -> dict[str, Any]:
    relative_key = normalize_relative_path(key, allow_empty=False, field="key")
    extension = PurePosixPath(relative_key).suffix.lower()
    if extension not in ALLOWED_EXTENSIONS:
        allowed = ", ".join(sorted(ALLOWED_EXTENSIONS))
        raise ToolError(f"Unsupported document type. Allowed extensions: {allowed}.")

    full_key = resolve_key(settings, relative_key)
    range_end = settings.max_document_bytes + 3
    try:
        response = client.get_object(
            Bucket=settings.bucket,
            Key=full_key,
            Range=f"bytes=0-{range_end}",
        )
        payload = response["Body"].read(settings.max_document_bytes + 4)
    except ClientError as exc:
        raise _safe_s3_error(exc) from exc

    return {
        "key": relative_key,
        "content": _decode_utf8_prefix(payload, settings.max_document_bytes),
        "truncated": len(payload) > settings.max_document_bytes,
    }


def _safe_s3_error(exc: ClientError) -> ToolError:
    code = str(exc.response.get("Error", {}).get("Code", ""))
    if code in {"404", "NoSuchKey", "NotFound"}:
        return ToolError("Document not found in the configured S3 prefix.")
    if code in {"403", "AccessDenied"}:
        return ToolError("S3 denied access. Check the runtime role and bucket policy.")
    return ToolError("The S3 request failed. Check the region, bucket, and runtime logs.")


mcp = FastMCP(
    "s3-document-helper",
    host="0.0.0.0",
    port=8000,
    streamable_http_path="/mcp",
    stateless_http=True,
)

READ_ONLY_ANNOTATIONS = ToolAnnotations(
    readOnlyHint=True,
    destructiveHint=False,
    idempotentHint=True,
    openWorldHint=True,
)


@mcp.tool(
    title="List S3 documents",
    description="List text documents beneath the configured S3 prefix.",
    annotations=READ_ONLY_ANNOTATIONS,
)
def list_documents(prefix: str = "", limit: int = 20) -> list[dict[str, Any]]:
    """List up to 50 documents. Prefixes are relative to S3_PREFIX."""
    settings = Settings.from_env()
    return list_documents_impl(prefix, limit, settings=settings, client=build_s3_client(settings))


@mcp.tool(
    title="Read an S3 document",
    description="Read one UTF-8 text document beneath the configured S3 prefix.",
    annotations=READ_ONLY_ANNOTATIONS,
)
def read_document(key: str) -> dict[str, Any]:
    """Read a document by the relative key returned by list_documents."""
    settings = Settings.from_env()
    return read_document_impl(key, settings=settings, client=build_s3_client(settings))


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

The important safety decisions are:

• callers provide keys relative to S3_PREFIX, never arbitrary bucket names;
• . and .. path segments and absolute keys are rejected;
• only .txt, .md, .json, .csv, .yaml, and .yml are accepted;
• reads stop at MAX_DOCUMENT_BYTES without splitting a UTF-8 character;
• AWS errors are translated into short tool errors without leaking raw service details; and
• the MCP annotations explicitly identify both tools as read-only and idempotent.

Create client.py for local and remote checks:

"""Small local or remote client for the S3 document MCP server."""

from __future__ import annotations

import asyncio
import json
import os
import sys

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


async def main() -> None:
    url = os.getenv("MCP_URL", "http://127.0.0.1:8000/mcp")
    token = os.getenv("BEARER_TOKEN", "").strip()
    headers = {"Authorization": f"Bearer {token}"} if token else {}

    async with streamablehttp_client(
        url,
        headers=headers,
        timeout=120,
        terminate_on_close=False,
    ) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("Tools:", ", ".join(tool.name for tool in tools.tools))

            listed = await session.call_tool("list_documents", {"prefix": "", "limit": 20})
            print(json.dumps(listed.structuredContent, indent=2, default=str))

            if len(sys.argv) > 1:
                document = await session.call_tool("read_document", {"key": sys.argv[1]})
                print(json.dumps(document.structuredContent, indent=2, default=str))


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

3. Test locally

Export the server configuration:

export AWS_REGION="us-west-2"
export S3_BUCKET="s3-document-mcp-${AWS_ACCOUNT_ID}-${AWS_REGION}"
export S3_PREFIX="documents/"
export MAX_DOCUMENT_BYTES="65536"

Start the server:

python server.py

AgentCore requires the process to listen at 0.0.0.0:8000/mcp. When you run the same process locally, use a trusted network and stop it when the test is complete.

From a second terminal, activate the environment, export the same variables, and call both tools:

source .venv/bin/activate
python client.py welcome.md

You should see:

Tools: list_documents, read_document

The first result lists welcome.md; the second returns its content with "truncated": false.

You can also inspect the server visually:

npx -y @modelcontextprotocol/inspector

Choose Streamable HTTP and connect to http://127.0.0.1:8000/mcp.

For automated tests, stub the S3 client rather than contacting AWS. The companion sample includes tests for pagination, limits, prefix containment, supported formats, UTF-8 boundaries, truncation, missing keys, and safe AWS error handling.

4. Create tutorial authentication with Cognito

The following quickstart uses Cognito's username/password authentication to obtain an access token. It is appropriate for this hands-on tutorial, not as the final sign-in flow for a public production application.

Create setup_cognito.sh:

#!/usr/bin/env bash
set -euo pipefail

: "${AWS_REGION:=us-west-2}"
: "${COGNITO_USERNAME:?Set COGNITO_USERNAME before sourcing this script.}"
: "${COGNITO_PASSWORD:?Set COGNITO_PASSWORD before sourcing this script.}"

command -v aws >/dev/null || { echo "AWS CLI is required." >&2; return 1 2>/dev/null || exit 1; }
command -v jq >/dev/null || { echo "jq is required." >&2; return 1 2>/dev/null || exit 1; }

export COGNITO_POOL_ID
COGNITO_POOL_ID="$(aws cognito-idp create-user-pool \
  --pool-name "s3-document-mcp-users" \
  --policies '{"PasswordPolicy":{"MinimumLength":12,"RequireUppercase":true,"RequireLowercase":true,"RequireNumbers":true,"RequireSymbols":true}}' \
  --region "$AWS_REGION" \
  --query 'UserPool.Id' \
  --output text)"

export COGNITO_CLIENT_ID
COGNITO_CLIENT_ID="$(aws cognito-idp create-user-pool-client \
  --user-pool-id "$COGNITO_POOL_ID" \
  --client-name "s3-document-mcp-client" \
  --no-generate-secret \
  --explicit-auth-flows ALLOW_USER_PASSWORD_AUTH ALLOW_REFRESH_TOKEN_AUTH \
  --region "$AWS_REGION" \
  --query 'UserPoolClient.ClientId' \
  --output text)"

aws cognito-idp admin-create-user \
  --user-pool-id "$COGNITO_POOL_ID" \
  --username "$COGNITO_USERNAME" \
  --message-action SUPPRESS \
  --region "$AWS_REGION" >/dev/null

aws cognito-idp admin-set-user-password \
  --user-pool-id "$COGNITO_POOL_ID" \
  --username "$COGNITO_USERNAME" \
  --password "$COGNITO_PASSWORD" \
  --permanent \
  --region "$AWS_REGION" >/dev/null

export BEARER_TOKEN
BEARER_TOKEN="$(aws cognito-idp initiate-auth \
  --client-id "$COGNITO_CLIENT_ID" \
  --auth-flow USER_PASSWORD_AUTH \
  --auth-parameters "USERNAME=$COGNITO_USERNAME,PASSWORD=$COGNITO_PASSWORD" \
  --region "$AWS_REGION" \
  --query 'AuthenticationResult.AccessToken' \
  --output text)"

export COGNITO_DISCOVERY_URL="https://cognito-idp.${AWS_REGION}.amazonaws.com/${COGNITO_POOL_ID}/.well-known/openid-configuration"

printf 'Cognito pool: %s\n' "$COGNITO_POOL_ID"
printf 'Client ID: %s\n' "$COGNITO_CLIENT_ID"
printf 'Discovery URL: %s\n' "$COGNITO_DISCOVERY_URL"
printf 'BEARER_TOKEN is exported in this shell and was not printed.\n'

Choose a temporary tutorial username and a strong password, then source the script so its exports remain in the current shell:

export AWS_REGION="us-west-2"
export COGNITO_USERNAME="tutorial-user"
read -s -p "Temporary Cognito password: " COGNITO_PASSWORD
export COGNITO_PASSWORD
printf '\n'
source ./setup_cognito.sh

The script deliberately does not print the access token. It exports:

• COGNITO_POOL_ID;
• COGNITO_CLIENT_ID;
• COGNITO_DISCOVERY_URL; and
• BEARER_TOKEN.

Access tokens expire. Re-run the initiate-auth command in the script when you need a fresh one.

For production, replace password authentication with Authorization Code plus PKCE or your organization's OIDC identity provider. Apply MFA, managed user lifecycle, appropriately short token lifetimes, and client-specific redirect URIs. AWS's Cognito authorization endpoint guide describes the browser-based OAuth flow.

5. Scaffold the AgentCore project

From the directory containing s3-document-mcp-source, create an MCP/CodeZip project:

cd ..
agentcore create \
  --name S3DocumentMCP \
  --protocol MCP \
  --build CodeZip

If your installed CLI opens the wizard, select Python, MCP, CodeZip, and no memory. The CLI generates agentcore/agentcore.json and an application directory; we replace the generated application code in the next step.

Copy the locally tested source into the generated application:

cp s3-document-mcp-source/server.py S3DocumentMCP/app/S3DocumentMCP/server.py
cp s3-document-mcp-source/client.py S3DocumentMCP/app/S3DocumentMCP/client.py
cp s3-document-mcp-source/requirements.txt S3DocumentMCP/app/S3DocumentMCP/requirements.txt
cd S3DocumentMCP

Update the generated runtime entry without disturbing the rest of the project:

python - <<'PY'
import json
import os
from pathlib import Path

path = Path("agentcore/agentcore.json")
config = json.loads(path.read_text())
runtime = config["runtimes"][0]
runtime.update({
    "name": "S3DocumentMCP",
    "build": "CodeZip",
    "entrypoint": "server.py",
    "codeLocation": "app/S3DocumentMCP/",
    "runtimeVersion": "PYTHON_3_12",
    "networkMode": "PUBLIC",
    "protocol": "MCP",
    "envVars": [
        {"name": "AWS_REGION", "value": os.environ["AWS_REGION"]},
        {"name": "S3_BUCKET", "value": os.environ["S3_BUCKET"]},
        {"name": "S3_PREFIX", "value": os.environ.get("S3_PREFIX", "documents/")},
        {"name": "MAX_DOCUMENT_BYTES", "value": "65536"},
    ],
    "instrumentation": {"enableOtel": True},
    "authorizerType": "CUSTOM_JWT",
    "authorizerConfiguration": {
        "customJwtAuthorizer": {
            "discoveryUrl": os.environ["COGNITO_DISCOVERY_URL"],
            "allowedClients": [os.environ["COGNITO_CLIENT_ID"]],
        }
    },
})
path.write_text(json.dumps(config, indent=2) + "\n")
PY

This follows the current agentcore.json runtime schema: CodeZip, Python 3.12, MCP protocol, runtime environment variables, OpenTelemetry, and a custom JWT authorizer.

Validate the project and preview its CloudFormation changes:

agentcore validate
cdk bootstrap "aws://${AWS_ACCOUNT_ID}/${AWS_REGION}"
agentcore deploy --dry-run

Review the dry run before creating anything.

6. Deploy and add least-privilege S3 access

Deploy the runtime:

agentcore deploy
agentcore status --json

The JSON status output identifies the runtime ARN, endpoint, CloudFormation stack, and execution-role ARN. Copy only the final role name from the execution-role ARN:

export RUNTIME_ROLE_NAME="REPLACE_WITH_RUNTIME_ROLE_NAME_FROM_AGENTCORE_STATUS"

Create s3-read-policy.json:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListOnlyTheDocumentPrefix",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::__S3_BUCKET__",
      "Condition": {
        "StringLike": {
          "s3:prefix": [
            "__S3_PREFIX__",
            "__S3_PREFIX__*"
          ]
        }
      }
    },
    {
      "Sid": "ReadOnlyDocuments",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::__S3_BUCKET__/__S3_PREFIX__*"
    }
  ]
}

Render the placeholders into a temporary policy and attach it as an inline policy:

python - <<'PY'
import os
from pathlib import Path

template = Path("s3-read-policy.json").read_text()
rendered = (
    template
    .replace("__S3_BUCKET__", os.environ["S3_BUCKET"])
    .replace("__S3_PREFIX__", os.environ.get("S3_PREFIX", "documents/"))
)
Path("/tmp/s3-document-mcp-policy.json").write_text(rendered)
PY

aws iam put-role-policy \
  --role-name "$RUNTIME_ROLE_NAME" \
  --policy-name "S3DocumentPrefixReadOnly" \
  --policy-document file:///tmp/s3-document-mcp-policy.json

This policy separates bucket listing from object reads and scopes both to documents/. It grants no PutObject, DeleteObject, ACL, bucket-policy, or cross-bucket permissions. For a long-lived production deployment, manage an explicit execution role in infrastructure as code and set its ARN as executionRoleArn, so manual policy drift cannot occur.

7. Call the deployed MCP server

Copy the AgentCore runtime ARN from agentcore status:

export AGENT_ARN="REPLACE_WITH_AGENTCORE_RUNTIME_ARN"
export ENCODED_AGENT_ARN="$(python -c 'import os, urllib.parse; print(urllib.parse.quote(os.environ["AGENT_ARN"], safe=""))')"
export MCP_URL="https://bedrock-agentcore.${AWS_REGION}.amazonaws.com/runtimes/${ENCODED_AGENT_ARN}/invocations?qualifier=DEFAULT"

The client already reads MCP_URL and BEARER_TOKEN, so invoke it directly:

python app/S3DocumentMCP/client.py welcome.md

The MCP SDK initializes the session, negotiates the protocol, lists the tools, and calls the two read-only operations. If the request returns 401 Unauthorized, refresh BEARER_TOKEN and confirm that the runtime's discovery URL and allowed client ID match the same Cognito user pool.

8. Observe and troubleshoot

Use the AgentCore CLI first:

agentcore logs
agentcore traces list
agentcore status --json

The runtime log group is also available under /aws/bedrock-agentcore/runtimes/ in Amazon CloudWatch.

Common failures:

• S3_BUCKET is required — the runtime environment variable is missing from agentcore.json.
• S3 denied access — the inline policy is missing, attached to the wrong role, or uses the wrong bucket/prefix.
• Document not found — the caller passed a key that was not returned by list_documents, or the object is outside S3_PREFIX.
• Unsupported document type — the key is not one of the allowed text formats.
• 401 from AgentCore — the access token expired or the JWT authorizer points to a different Cognito pool/client.
• Import failure for mcp.server.fastmcp — MCP 2.x was installed despite the compatibility constraint.

Production hardening

This tutorial is intentionally small. Before production use:

• create the runtime role and policy in CDK, CloudFormation, or Terraform;
• use Authorization Code with PKCE or enterprise OIDC instead of password authentication;
• restrict which users and clients may invoke the runtime;
• consider a VPC only when the server must reach private resources;
• add CloudWatch alarms for authentication failures, tool errors, latency, and throttling;
• add object-level classification rules before returning sensitive text;
• treat tool descriptions and returned document content as untrusted input to the model; and
• add explicit confirmation and audit controls before introducing any future mutation tool.

Cleanup

Cleanup is destructive. Confirm that the variables still refer only to this tutorial before running it.

Delete the manually attached policy first so it cannot block CloudFormation from deleting the role:

aws iam delete-role-policy \
  --role-name "$RUNTIME_ROLE_NAME" \
  --policy-name "S3DocumentPrefixReadOnly"

Remove the AgentCore resources from the project configuration and deploy the removal:

agentcore remove all
agentcore deploy

Delete the sample object, bucket, and Cognito pool:

aws s3 rm "s3://${S3_BUCKET}" --recursive
aws s3api delete-bucket --bucket "$S3_BUCKET" --region "$AWS_REGION"
aws cognito-idp delete-user-pool \
  --user-pool-id "$COGNITO_POOL_ID" \
  --region "$AWS_REGION"

unset BEARER_TOKEN COGNITO_PASSWORD COGNITO_CLIENT_ID COGNITO_DISCOVERY_URL COGNITO_POOL_ID

Finally, run agentcore status and inspect the CloudFormation stack to confirm teardown completed.

What we built

We now have a remote MCP server with a deliberately narrow security boundary:

• Cognito authenticates the caller;
• AgentCore Runtime hosts the MCP transport and runtime;
• IAM limits the server to one S3 prefix;
• FastMCP exposes two read-only tools; and
• CloudWatch provides operational visibility.

That is a useful foundation for document lookup without giving an AI client the ability to mutate the bucket.

References

• Deploy MCP servers in Amazon Bedrock AgentCore Runtime
• AgentCore Runtime concepts
• AgentCore CLI configuration reference
• MCP Streamable HTTP transport
• AWS Architecture Icons
• IAM policies for Amazon S3