AgentCore 런타임에 MCP 서버 배포
Amazon Bedrock AgentCore 런타임을 사용하면 AgentCore 런타임에서 모델 컨텍스트 프로토콜(MCP) 서버를 배포하고 실행할 수 있습니다. 이 가이드에서는 첫 번째 MCP 서버를 생성, 테스트 및 배포하는 방법을 안내합니다.
예제는 https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
이 섹션에서는 다음을 배웁니다.
-
도구를 사용하여 MCP 서버를 생성하는 방법
-
로컬에서 서버를 테스트하는 방법
-
에 서버를 배포하는 방법 AWS
-
배포된 서버를 호출하는 방법
MCP에 대한 자세한 내용은 MCP 프로토콜 계약을 참조하세요.
주제
Amazon Bedrock AgentCore가 MCP를 지원하는 방법
MCP 프로토콜로 Amazon Bedrock AgentCore 런타임을 구성하면 서비스는 대부분의 공식 MCP 서버 SDK에서 지원하는 기본 경로인 경로 0.0.0.0:8000/mcp에서 MCP 서버 컨테이너를 사용할 수 있을 것으로 예상합니다. SDKs
Amazon Bedrock AgentCore는 상태 비저장 및 상태 저장 스트리밍 가능 HTTP MCP 서버를 모두 지원합니다. 기본적으로 상태 비저장 모드(stateless_http=True)는 기본 MCP 서버에 권장됩니다. 플랫폼은 요청 없이 모든 요청에 대한 Mcp-Session-Id 헤더를 자동으로 추가하므로 MCP 클라이언트는 동일한 Amazon Bedrock AgentCore 런타임 세션에 대한 연결 연속성을 유지할 수 있습니다.
멀티턴 상호 작용(추출), LLM 생성 콘텐츠(샘플링) 또는 진행률 알림이 필요한 MCP 서버의 경우 상태 저장 모드(stateless_http=False)를 사용하면 이러한 기능을 사용할 수 있습니다. 상태 저장 모드에서 런타임은 동일한 호출 내의 요청 간에 MCP 세션 상태를 보존합니다. 자세한 내용은 상태 저장 MCP 서버 기능을 참조하세요.
InvokeAgentRuntime API의 페이로드는 직접 전달되므로 MCP와 같은 프로토콜의 RPC 메시지를 쉽게 프록시할 수 있습니다.
사전 조건
-
Python 3.10 이상 설치 및 Python에 대한 기본 이해
-
적절한 권한과 로컬 자격 증명이 구성된 AWS 계정
1단계: MCP 서버 생성
필수 패키지 설치
먼저 MCP 패키지를 설치합니다.
pip install mcp
첫 번째 MCP 서버 생성
라는 새 파일을 생성합니다my_mcp_server.py.
# my_mcp_server.py from mcp.server.fastmcp import FastMCP from starlette.responses import JSONResponse mcp = FastMCP(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." if __name__ == "__main__": mcp.run(transport="streamable-http")
코드 이해
-
FastMCP : 도구를 호스팅할 수 있는 MCP 서버를 생성합니다.
-
@mcp.tool() : Python 함수를 MCP 도구로 변환하는 데코레이터
-
도구: 다양한 유형의 작업을 보여주는 세 가지 간단한 도구
-
stateless_http=True: 기본 MCP 서버의 기본값인 상태 비저장 모드로 서버를 구성합니다.
작은 정보
멀티턴 상호 작용(추출) 또는 LLM 생성 콘텐츠(샘플링)가 필요한 MCP 서버의 경우 stateless_http=False를 사용하여 상태 저장 모드를 활성화합니다. 상태 저장 MCP 서버는 동일한 도구 호출 내에서 여러 요청의 세션 컨텍스트를 유지합니다. 자세한 내용은 상태 저장 MCP 서버 기능을 참조하세요.
2단계: 로컬에서 MCP 서버 테스트
MCP 서버 시작
MCP 서버를 로컬에서 실행합니다.
python my_mcp_server.py
서버가 포트에서 실행 중임을 나타내는 출력이 표시되어야 합니다8000.
MCP 클라이언트로 테스트
새 터미널에서 새 파일을 생성하고를 사용하여 my_mcp_client.py 실행합니다. python my_mcp_client.py
# my_mcp_client.py import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): mcp_url = "http://localhost:8000/mcp" headers = {} async with streamablehttp_client(mcp_url, headers, timeout=120, terminate_on_close=False) as ( read_stream, write_stream, _, ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tool_result = await session.list_tools() print(tool_result) asyncio.run(main())
MCP 인스펙터를 사용한 로컬 테스트에 설명된 대로 MCP 인스펙터를 사용하여 서버를 테스트할 수도 있습니다.
3단계:에 MCP 서버 배포 AWS
배포 도구 설치
AgentCore CLI를 설치합니다.
npm install -g @aws/agentcore
AgentCore CLI를 사용하여 에이전트를 AgentCore 런타임에 배포합니다.
다음 구조의 프로젝트 폴더를 생성합니다.
## Project Folder Structure your_project_directory/ ├── mcp_server.py # Your main agent code ├── requirements.txt # Dependencies for your agent └── __init__.py # Makes the directory a Python package
라는 새 파일을 생성하고 여기에 다음을 requirements.txt 추가합니다.
mcp
requirements.txt는 에이전트가 AgentCore 런타임에 배포하는 데 필요한 요구 사항을 지정합니다.
배포를 위한 프로젝트 생성
프로젝트를 생성하기 전에 인증을 위한 Cognito 사용자 풀 설정에 설명된 대로 인증을 위한 Cognito 사용자 풀을 설정해야 합니다. 이렇게 하면 배포된 서버에 대한 보안 액세스에 필요한 OAuth 토큰이 제공됩니다.
참고
2025년 10월 7일부터 Amazon Bedrock AgentCore는 OAuth 인증을 사용할 때 워크로드 자격 증명 권한에 서비스 연결 역할을 사용합니다. 이 변경에 대한 자세한 내용은 자격 증명 서비스 연결 역할을 참조하세요.
인증을 설정한 후 MCP 프로토콜을 사용하여 새 프로젝트를 스캐폴드합니다.
agentcore create --protocol MCP
대화형 프롬프트에 따라 프로젝트 이름을 제공합니다. CLI는 agentcore/agentcore.json 구성 파일을 포함하여 프로젝트 구조를 스캐폴드합니다. 생성된 프로젝트의 에이전트 코드 디렉터리에 my_mcp_server.py 파일을 복사하고의 진입점이 서버 파일을 agentcore/agentcore.json 가리키는지 확인합니다.
에 배포 AWS
에이전트를 배포합니다.
agentcore deploy
이 명령은 다음을 수행합니다.
-
에이전트 코드 및 종속성 패키징
-
Amazon S3에 배포 아티팩트 업로드
-
Amazon Bedrock AgentCore 런타임 생성
-
에 에이전트 배포 AWS
배포 후 다음과 같은 에이전트 런타임 ARN을 받게 됩니다.
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
4단계: 배포된 MCP 서버 호출
MCP 클라이언트로 테스트(원격)
테스트하기 전에 다음 환경 변수를 설정합니다.
-
에이전트 ARN을 환경 변수로 내보냅니다.
export AGENT_ARN="agent_arn" -
베어러 토큰을 환경 변수로 내보냅니다.
export BEARER_TOKEN="bearer_token"
Accept 헤더를 전달하는 경우 MCPapplication/json 및 입니다text/event-stream.
새 파일을 생성하고를 사용하여 my_mcp_client_remote.py 실행합니다. python my_mcp_client_remote.py
import asyncio import os import sys from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): agent_arn = os.getenv('AGENT_ARN') bearer_token = os.getenv('BEARER_TOKEN') if not agent_arn or not bearer_token: print("Error: AGENT_ARN or BEARER_TOKEN environment variable is not set") sys.exit(1) encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F') mcp_url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT" headers = {"authorization": f"Bearer {bearer_token}","Content-Type":"application/json"} print(f"Invoking: {mcp_url}, \nwith headers: {headers}\n") async with streamablehttp_client(mcp_url, headers, timeout=120, terminate_on_close=False) as ( read_stream, write_stream, _, ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tool_result = await session.list_tools() print(tool_result) asyncio.run(main())
MCP 인스펙터를 사용한 원격 테스트에 설명된 대로 MCP 인스펙터를 사용하여 배포된 서버를 테스트할 수도 있습니다.
OAuth 구성 에이전트에 대한 인증 오류 응답
OAuth로 구성된 에이전트는 RFC 6749(OAuth 2.0)
401 무단 - 인증 누락
권한 부여 헤더에 베어러 토큰이 제공되지 않은 경우 응답은 다음과 같습니다.
HTTP/1.1 401 Unauthorized WWW-Authenticate: Bearer resource_metadata="https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{ESCAPED_ARN}/invocations/.well-known/oauth-protected-resource?qualifier={QUALIFIER}"
Auth0을 사용한 엔드 투 엔드 흐름
이 섹션에서는 Auth0을 자격 증명 공급자로 사용하는 OAuth Auth0 인증을 보여줍니다. 이 예제에서는 Auth0을 사용합니다. 동적 클라이언트 등록(DCR)을 지원하므로 클라이언트가 런타임에 프로그래밍 방식으로 자신을 등록할 수 있으므로 클라이언트 설정 프로세스가 간소화됩니다.
1단계 - 3단계: MCP 서버 생성 및 테스트
1단계: MCP 서버 생성부터 3단계:에 MCP 서버 배포까지 1~3 AWS단계를 수행하여 MCP 서버를 생성하고 테스트합니다.
4단계: Auth0 애플리케이션 생성
Okta의 Auth0에서 Auth0 설정 지침을 따릅니다. Auth0
동적 클라이언트 등록 활성화:
-
대시보드 → 설정 → 고급
-
"OIDC Dynamic Application Registration" → ON으로 전환
-
변경 내용을 저장합니다.
자세한 내용은 Auth0 동적 클라이언트 등록 설명서를
5단계: 배포를 위한 프로젝트 생성
인증을 설정한 후 MCP 프로토콜을 사용하여 새 프로젝트를 스캐폴드합니다.
agentcore create --protocol MCP
대화형 프롬프트에 따라 프로젝트 이름을 제공합니다. CLI는 agentcore/agentcore.json 구성 파일을 포함하여 프로젝트 구조를 스캐폴드합니다. 생성된 프로젝트의 에이전트 코드 디렉터리에 my_mcp_server.py 파일을 복사하고의 진입점이 서버 파일을 agentcore/agentcore.json 가리키는지 확인합니다.
6단계:에 배포 AWS
에이전트를 배포합니다.
agentcore deploy
이 명령은 다음을 수행합니다.
-
에이전트 코드 및 종속성 패키징
-
Amazon S3에 배포 아티팩트 업로드
-
Amazon Bedrock AgentCore 런타임 생성
-
에 에이전트 배포 AWS
배포 후 다음과 같은 에이전트 런타임 ARN을 받게 됩니다.
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
7단계: 배포된 에이전트 호출
이 클라이언트는 Auth0-specific0별 수정 사항이 포함된 공식 MCP SDK simple-auth-client 예제
참고
동적 클라이언트 등록과 함께 Auth0을 사용하는 경우 JWT 토큰을 수신하려면 권한 부여 요청에 audience 파라미터를 포함해야 합니다. 이 파라미터가 없으면 Auth0은 표준 JWT 토큰 대신 불투명 토큰 또는 JWE(암호화된) 토큰을 반환합니다. MCP SDK는 OAuth 2.0의 resource 파라미터(RFC 8707)를 전송하지만 Auth0에는 JWT 토큰에 대한 OIDC audience 파라미터가 필요합니다. 두 파라미터 모두 비슷한 목적을 수행하지만 Auth0은 audience 우선 순위를 지정합니다. 자세한 내용은 Auth0 커뮤니티 - 동적 애플리케이션 등록을 사용하는 JWT 토큰을 참조하세요
다음 코드가 포함된 mcp_auth0_client.py라는 파일을 생성합니다. 이 클라이언트는 대상 파라미터를 포함한 Auth0-specific 요구 사항을 처리합니다.
참고
이 코드에는 모든 HTTP 요청에 User-Agent 헤더를 주입하기 위한 httpx 패치가 포함되어 있습니다. 이는 MCP Python SDK가 현재 HTTP 요청에 User-Agent 헤더를 포함하지 않기 때문에 필요합니다. 이로 인해 User-Agent 헤더가 필요한 AWS WAF 규칙에 문제가 발생할 수 있습니다. 자세한 내용은 MCP Python SDK 문제 #1664
#!/usr/bin/env python3 """ MCP client with OAuth authentication support for Auth0. Based on the official MCP SDK simple-auth-client example with Auth0 compatibility. Adds support for Auth0's 'audience' parameter requirement. Usage: # Required export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" # Required for Auth0 export AUTH0_API_IDENTIFIER="your-api-identifier" # Optional - custom endpoint for beta/dev environments export CUSTOM_ENDPOINT="https://beta.example.com" python mcp_auth0_client.py The client will automatically: - Encode the Agent ARN for use in the URL - Construct the MCP invocation endpoint URL - Add Auth0 'audience' parameter to authorization requests (when using Auth0) - Work with any OAuth 2.0 compliant identity provider """ import asyncio import httpx import os import threading import time import webbrowser from datetime import timedelta from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any from urllib.parse import parse_qs, urlencode, urlparse, urlunparse # Patch httpx at the request level to inject User-Agent header # This ensures ALL HTTP requests have the User-Agent header, including OAuth discovery calls _original_httpx_request = httpx.Request.__init__ def _patched_httpx_request_init(self, method, url, *args, **kwargs): """Patched Request.__init__ that injects User-Agent header into all HTTP requests.""" # Get or create headers headers = kwargs.get('headers') if headers is None: headers = {} kwargs['headers'] = headers # Convert to mutable dict if needed if not isinstance(headers, dict): headers = dict(headers) kwargs['headers'] = headers # Inject User-Agent if not present (case-insensitive check) if 'User-Agent' not in headers and 'user-agent' not in headers: headers['User-Agent'] = 'python-mcp-sdk/1.0 (BedrockAgentCore-Runtime)' # Call original __init__ _original_httpx_request(self, method, url, *args, **kwargs) # Apply the patch globally before importing MCP modules httpx.Request.__init__ = _patched_httpx_request_init # Now import MCP modules - they will use patched httpx from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession from mcp.client.sse import sse_client from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken class InMemoryTokenStorage(TokenStorage): """Simple in-memory token storage implementation.""" def __init__(self): self._tokens: OAuthToken | None = None self._client_info: OAuthClientInformationFull | None = None async def get_tokens(self) -> OAuthToken | None: return self._tokens async def set_tokens(self, tokens: OAuthToken) -> None: self._tokens = tokens async def get_client_info(self) -> OAuthClientInformationFull | None: return self._client_info async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: self._client_info = client_info class CallbackHandler(BaseHTTPRequestHandler): """Simple HTTP handler to capture OAuth callback.""" def __init__(self, request, client_address, server, callback_data): """Initialize with callback data storage.""" self.callback_data = callback_data super().__init__(request, client_address, server) def do_GET(self): """Handle GET request from OAuth redirect.""" parsed = urlparse(self.path) query_params = parse_qs(parsed.query) if "code" in query_params: self.callback_data["authorization_code"] = query_params["code"][0] self.callback_data["state"] = query_params.get("state", [None])[0] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b""" <html> <body> <h1>Authorization Successful!</h1> <p>You can close this window and return to the terminal.</p> <script>setTimeout(() => window.close(), 2000);</script> </body> </html> """) elif "error" in query_params: self.callback_data["error"] = query_params["error"][0] self.send_response(400) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write( f""" <html> <body> <h1>Authorization Failed</h1> <p>Error: {query_params["error"][0]}</p> <p>You can close this window and return to the terminal.</p> </body> </html> """.encode() ) else: self.send_response(404) self.end_headers() def log_message(self, format, *args): """Suppress default logging.""" pass class CallbackServer: """Simple server to handle OAuth callbacks.""" def __init__(self, port=3030): self.port = port self.server = None self.thread = None self.callback_data = {"authorization_code": None, "state": None, "error": None} def _create_handler_with_data(self): """Create a handler class with access to callback data.""" callback_data = self.callback_data class DataCallbackHandler(CallbackHandler): def __init__(self, request, client_address, server): super().__init__(request, client_address, server, callback_data) return DataCallbackHandler def start(self): """Start the callback server in a background thread.""" handler_class = self._create_handler_with_data() self.server = HTTPServer(("localhost", self.port), handler_class) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() print(f"🖥️ Started callback server on http://localhost:{self.port}") def stop(self): """Stop the callback server.""" if self.server: self.server.shutdown() self.server.server_close() if self.thread: self.thread.join(timeout=1) def wait_for_callback(self, timeout=300): """Wait for OAuth callback with timeout.""" start_time = time.time() while time.time() - start_time < timeout: if self.callback_data["authorization_code"]: return self.callback_data["authorization_code"] elif self.callback_data["error"]: raise Exception(f"OAuth error: {self.callback_data['error']}") time.sleep(0.1) raise Exception("Timeout waiting for OAuth callback") def get_state(self): """Get the received state parameter.""" return self.callback_data["state"] def add_auth0_audience_parameter(authorization_url: str, audience: str) -> str: """ Add Auth0 'audience' parameter to authorization URL. Auth0 requires the 'audience' parameter to identify which API's token settings to use. Without it, Auth0 returns opaque tokens or JWE instead of JWT. This function properly adds the audience parameter while preserving all existing query parameters (including the OAuth 'resource' parameter). Args: authorization_url: The authorization URL from the OAuth flow audience: The Auth0 API identifier (e.g., "runtime-api") Returns: Modified URL with audience parameter added Reference: https://auth0.com/docs/secure/tokens/access-tokens/get-access-tokens """ # Only apply to Auth0 URLs that don't already have audience if 'auth0.com' not in authorization_url or 'audience=' in authorization_url: return authorization_url # Parse URL and query parameters parsed = urlparse(authorization_url) query_params = parse_qs(parsed.query, keep_blank_values=True) # Add audience parameter query_params['audience'] = [audience] # Rebuild URL with new parameter new_query = urlencode(query_params, doseq=True) return urlunparse(( parsed.scheme, parsed.netloc, parsed.path, parsed.params, new_query, parsed.fragment )) class SimpleAuthClient: """Simple MCP client with Auth0 OAuth support.""" def __init__( self, server_url: str, transport_type: str = "streamable-http", auth0_audience: str | None = None, ): self.server_url = server_url self.transport_type = transport_type self.auth0_audience = auth0_audience self.session: ClientSession | None = None async def connect(self): """Connect to the MCP server.""" print(f"🔗 Attempting to connect to {self.server_url}...") try: callback_server = CallbackServer(port=3030) callback_server.start() async def callback_handler() -> tuple[str, str | None]: """Wait for OAuth callback and return auth code and state.""" print("⏳ Waiting for authorization callback...") try: auth_code = callback_server.wait_for_callback(timeout=300) return auth_code, callback_server.get_state() finally: callback_server.stop() client_metadata_dict = { "client_name": "MCP Auth0 Client", "redirect_uris": ["http://localhost:3030/callback"], "grant_types": ["authorization_code", "refresh_token"], "response_types": ["code"], } async def redirect_handler(authorization_url: str) -> None: """Redirect handler that opens the URL in a browser with Auth0 audience parameter.""" # Add Auth0 audience parameter if configured if self.auth0_audience: authorization_url = add_auth0_audience_parameter( authorization_url, self.auth0_audience ) webbrowser.open(authorization_url) print("\n🔧 Creating OAuth client provider...") # Create OAuth authentication handler # Note: httpx.AsyncClient is globally patched to inject User-Agent header oauth_auth = OAuthClientProvider( server_url=self.server_url, client_metadata=OAuthClientMetadata.model_validate(client_metadata_dict), storage=InMemoryTokenStorage(), redirect_handler=redirect_handler, callback_handler=callback_handler, ) print("🔧 OAuth client provider created successfully") # Create transport with auth handler based on transport type if self.transport_type == "sse": print("📡 Opening SSE transport connection with auth...") async with sse_client( url=self.server_url, auth=oauth_auth, timeout=60, ) as (read_stream, write_stream): await self._run_session(read_stream, write_stream, None) else: print("📡 Opening StreamableHTTP transport connection with auth...") async with streamablehttp_client( url=self.server_url, auth=oauth_auth, timeout=timedelta(seconds=60), ) as (read_stream, write_stream, get_session_id): await self._run_session(read_stream, write_stream, get_session_id) except Exception as e: print(f"❌ Failed to connect: {e}") import traceback traceback.print_exc() async def _run_session(self, read_stream, write_stream, get_session_id): """Run the MCP session with the given streams.""" print("🤝 Initializing MCP session...") async with ClientSession(read_stream, write_stream) as session: self.session = session print("⚡ Starting session initialization...") await session.initialize() print("✨ Session initialization complete!") print(f"\n✅ Connected to MCP server at {self.server_url}") if get_session_id: session_id = get_session_id() if session_id: print(f"Session ID: {session_id}") # Run interactive loop await self.interactive_loop() async def list_tools(self): """List available tools from the server.""" if not self.session: print("❌ Not connected to server") return try: result = await self.session.list_tools() if hasattr(result, "tools") and result.tools: print("\n📋 Available tools:") for i, tool in enumerate(result.tools, 1): print(f"{i}. {tool.name}") if tool.description: print(f" Description: {tool.description}") print() else: print("No tools available") except Exception as e: print(f"❌ Failed to list tools: {e}") async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None = None): """Call a specific tool.""" if not self.session: print("❌ Not connected to server") return try: result = await self.session.call_tool(tool_name, arguments or {}) print(f"\n🔧 Tool '{tool_name}' result:") if hasattr(result, "content"): for content in result.content: if content.type == "text": print(content.text) else: print(content) else: print(result) except Exception as e: print(f"❌ Failed to call tool '{tool_name}': {e}") async def interactive_loop(self): """Run interactive command loop.""" print("\n🎯 Interactive MCP Client") print("Commands:") print(" list - List available tools") print(" call <tool_name> [args] - Call a tool") print(" quit - Exit the client") print() while True: try: command = input("mcp> ").strip() if not command: continue if command == "quit": break elif command == "list": await self.list_tools() elif command.startswith("call "): parts = command.split(maxsplit=2) tool_name = parts[1] if len(parts) > 1 else "" if not tool_name: print("❌ Please specify a tool name") continue # Parse arguments (simple JSON-like format) arguments = {} if len(parts) > 2: import json try: arguments = json.loads(parts[2]) except json.JSONDecodeError: print("❌ Invalid arguments format (expected JSON)") continue await self.call_tool(tool_name, arguments) else: print("❌ Unknown command. Try 'list', 'call <tool_name>', or 'quit'") except KeyboardInterrupt: print("\n\n👋 Goodbye!") break except EOFError: break async def main(): """Main entry point.""" # Get Agent ARN from environment agent_arn = os.getenv("AGENT_ARN") if not agent_arn: print("❌ Please set AGENT_ARN environment variable") print("Example: export AGENT_ARN='arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234'") return # Encode the ARN for use in URL encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F') # Get base URL - use custom endpoint or default to production base_endpoint = os.getenv("CUSTOM_ENDPOINT", "https://bedrock-agentcore.us-west-2.amazonaws.com") # Construct MCP URL from encoded ARN (no qualifier - SDK discovers it from PRM API) server_url = f"{base_endpoint}/runtimes/{encoded_arn}/invocations" # Get Auth0 configuration (required only for Auth0) auth0_audience = os.getenv("AUTH0_API_IDENTIFIER") # Get optional transport type transport_type = os.getenv("MCP_TRANSPORT_TYPE", "streamable-http") print("🚀 MCP Auth0 Client") print(f"Agent ARN: {agent_arn}") print(f"Endpoint: {base_endpoint}") print(f"Connecting to: {server_url}") print(f"Transport type: {transport_type}") if auth0_audience: print(f"Auth0 audience: {auth0_audience}") # Start connection flow - OAuth will be handled automatically client = SimpleAuthClient( server_url, transport_type, auth0_audience, ) await client.connect() def cli(): """CLI entry point for uv script.""" asyncio.run(main()) if __name__ == "__main__": cli()
클라이언트를 사용하려면:
-
필요한 환경 변수 설정:
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Auth0-specific 환경 변수를 설정합니다(Auth0에만 필요).
export AUTH0_API_IDENTIFIER="your-api-identifier" -
클라이언트를 실행합니다.
python mcp_auth0_client.py
클라이언트는 자동으로 다음을 수행합니다.
-
URL에 사용할 에이전트 ARN 인코딩
-
MCP 호출 엔드포인트 URL 구성
-
권한 부여 요청에 Auth0
audience파라미터 추가(Auth0 사용 시) -
OAuth 2.0 호환 자격 증명 공급자 작업
부록
인증을 위한 Cognito 사용자 풀 설정
새 파일을 setup_cognito.sh 생성하고 다음 콘텐츠를 추가합니다.
#!/bin/bash # Create User Pool and capture Pool ID directly export POOL_ID=$(aws cognito-idp create-user-pool \ --pool-name "MyUserPool" \ --policies '{"PasswordPolicy":{"MinimumLength":8}}' \ --region $REGION | jq -r '.UserPool.Id') # Create App Client and capture Client ID directly export CLIENT_ID=$(aws cognito-idp create-user-pool-client \ --user-pool-id $POOL_ID \ --client-name "MyClient" \ --no-generate-secret \ --explicit-auth-flows "ALLOW_USER_PASSWORD_AUTH" "ALLOW_REFRESH_TOKEN_AUTH" \ --region $REGION | jq -r '.UserPoolClient.ClientId') # Create User aws cognito-idp admin-create-user \ --user-pool-id $POOL_ID \ --username $USERNAME \ --region $REGION \ --message-action SUPPRESS > /dev/null # Set Permanent Password aws cognito-idp admin-set-user-password \ --user-pool-id $POOL_ID \ --username $USERNAME \ --password $PASSWORD \ --region $REGION \ --permanent > /dev/null # Authenticate User and capture Access Token export BEARER_TOKEN=$(aws cognito-idp initiate-auth \ --client-id "$CLIENT_ID" \ --auth-flow USER_PASSWORD_AUTH \ --auth-parameters USERNAME=$USERNAME,PASSWORD=$PASSWORD \ --region $REGION | jq -r '.AuthenticationResult.AccessToken') # Output the required values echo "Pool id: $POOL_ID" echo "Discovery URL: https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" echo "Client ID: $CLIENT_ID" echo "Bearer Token: $BEARER_TOKEN"
터미널 창을 열고 다음 환경 변수를 설정합니다.
-
리전- 사용하려는 AWS 리전 -
USERNAME- 새 사용자의 사용자 이름 -
암호- 새 사용자의 암호
export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD
명령을 사용하여 스크립트를 실행합니다source setup_cognito.sh.
참고
자세한 OAuth 인증 설정 및 서비스 연결 역할 정보는 인바운드 인증 및 아웃바운드 인증을 사용하여 인증 및 권한 부여를 참조하세요.
이 스크립트를 실행한 후 배포 구성에 사용할 다음 값을 기록해 둡니다.
-
검색 URL:
agentcore create단계에서 사용됨 -
클라이언트 ID:
agentcore create단계에서 사용됨 -
베어러 토큰: 배포된 서버를 호출할 때 사용됩니다.
MCP 검사기를 사용한 로컬 테스트
MCP Inspector는 MCP 서버를 테스트하기 위한 시각적 도구입니다. 이를 사용하려면 다음이 필요합니다.
-
Node.js 및 npm 설치됨
MCP Inspector를 설치하고 실행합니다.
npx @modelcontextprotocol/inspector
이렇게 하면 다음이 수행됩니다.
-
MCP Inspector 서버 시작
-
터미널에 URL 표시(일반적으로
http://localhost:6274)
Inspector를 사용하려면:
-
브라우저
http://localhost:6274에서 로 이동 -
MCP 서버 URL(
http://localhost:8000/mcp)을 MCP Inspector 연결 필드에 붙여넣습니다. -
사이드바에 도구가 나열됩니다.
-
도구를 클릭하여 테스트합니다.
-
파라미터를 입력합니다(예:
add_numbers의 경우a및b의 값 입력). -
"Call Tool"을 클릭하여 결과를 확인합니다.
MCP 검사기를 사용한 원격 테스트
MCP Inspector를 사용하여 배포된 서버를 테스트할 수도 있습니다. 먼저 에이전트 ARN을 URL로 인코딩합니다.
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
그러면 URL로 인코딩된 ARN이 출력됩니다.
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
그런 다음 MCP Inspector에 연결합니다.
-
MCP Inspector를 시작합니다.
npx @modelcontextprotocol/inspector -
웹 인터페이스에서:
-
전송으로 "스트리밍 가능한 HTTP"를 선택합니다.
-
인코딩된 ARN을 사용하여 에이전트의 엔드포인트 URL을 입력합니다. 에이전트의 ARN과 동일한 리전을 사용해야 합니다.
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTus-west-2의 예:
https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123/invocations?qualifier=DEFAULT -
헤더 이름
Authorization및 값을 사용하여 인증 섹션에 베어러 토큰 추가Bearer YOUR_TOKEN -
"Connect"를 클릭합니다.
-
-
로컬에서 했던 것처럼 도구를 테스트합니다.
