Recevoir une invite depuis une AgentCore passerelle
Pour obtenir une invite spécifique, envoyez une requête POST au point de terminaison MCP de la passerelle et spécifiez prompts/get comme méthode dans le corps de la demande, le nom de l'invite et les arguments :
POST /mcp HTTP/1.1
Host: ${GatewayEndpoint}
Content-Type: application/json
Authorization: ${Authorization header}
${RequestBody}
Remplacez les valeurs suivantes :
-
${GatewayEndpoint}— L'URL de la passerelle, telle que fournie dans la réponse de l'CreateGatewayAPI.
-
${Authorization header}— Les informations d'autorisation fournies par le fournisseur d'identité lorsque vous configurez l'autorisation entrante.
-
${RequestBody}— La charge utile JSON du corps de la requête, telle que spécifiée dans la section Obtenir une invite dans le Model Context Protocol (MCP). Incluez prompts/get en tant que method et incluez le name de l'invite et sesarguments.
La réponse renvoie l'invite rendue sous la forme d'un tableau de messages, chacun ayant un rôle et un contenu.
L'prompts/getopération transmet la demande en direct au serveur MCP en aval. Le nom de l'invite doit inclure le préfixe cible (par exemple,myTarget___myPrompt).
Exemples de code pour obtenir une invite
Pour voir des exemples de réception d'une invite depuis la passerelle, sélectionnez l'une des méthodes suivantes :
Exemple
- curl
-
-
La requête curl suivante montre un exemple de demande pour qu'une invite soit appelée myTarget___code_review via une passerelle avec l'IDmygateway-abcdefghij.
curl -X POST \
https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-d '{
"jsonrpc": "2.0",
"id": "get-prompt-request",
"method": "prompts/get",
"params": {
"name": "myTarget___code_review",
"arguments": {
"language": "python",
"code": "print(\"hello\")"
}
}
}'
- Python requests package
-
-
import requests
import json
def get_prompt(gateway_url, access_token, prompt_name, arguments):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
payload = {
"jsonrpc": "2.0",
"id": "get-prompt-request",
"method": "prompts/get",
"params": {
"name": prompt_name,
"arguments": arguments
}
}
response = requests.post(gateway_url, headers=headers, json=payload)
return response.json()
# Example usage
gateway_url = "https://${GatewayEndpoint}/mcp" # Replace with your actual gateway endpoint
access_token = "${AccessToken}" # Replace with your actual access token
result = get_prompt(
gateway_url,
access_token,
"myTarget___code_review", # Replace with {targetName}___{promptName}
{"language": "python", "code": "print('hello')"}
)
print(json.dumps(result, indent=2))
- MCP Client
-
-
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio
async def execute_mcp(
url,
token,
prompt_name,
prompt_arguments,
headers=None
):
default_headers = {
"Authorization": f"Bearer {token}"
}
headers = {**default_headers, **(headers or {})}
async with streamablehttp_client(
url=url,
headers=headers,
) as (
read_stream,
write_stream,
callA,
):
async with ClientSession(read_stream, write_stream) as session:
# 1. Perform initialization handshake
print("Initializing MCP...")
_init_response = await session.initialize()
print(f"MCP Server Initialize successful! - {_init_response}")
# 2. Get specific prompt
print(f"Getting prompt: {prompt_name}")
prompt_response = await session.get_prompt(
name=prompt_name,
arguments=prompt_arguments
)
print(f"Prompt response: {prompt_response}")
return prompt_response
async def main():
url = "https://${GatewayEndpoint}/mcp"
token = "your_bearer_token_here"
prompt_name = "myTarget___code_review"
prompt_arguments = {
"language": "python",
"code": "print('hello')"
}
await execute_mcp(
url=url,
token=token,
prompt_name=prompt_name,
prompt_arguments=prompt_arguments
)
if __name__ == "__main__":
asyncio.run(main())
- LangGraph MCP Client
-
-
REMARQUE : le support rapide de l'adaptateur LangGraph MCP peut varier. Utilisez l'approche client MCP ci-dessus pour une prompts/get mise en œuvre la plus fiable possible.
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def get_prompt(url, token, prompt_name, arguments):
headers = {"Authorization": f"Bearer {token}"}
async with streamablehttp_client(url=url, headers=headers) as (
read_stream, write_stream, callA
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
response = await session.get_prompt(
name=prompt_name,
arguments=arguments
)
for message in response.messages:
print(f"{message.role}: {message.content}")
asyncio.run(get_prompt(
"https://${GatewayEndpoint}/mcp",
"${AccessToken}",
"myTarget___code_review",
{"language": "python", "code": "print('hello')"}
))
Erreurs
L'prompts/getopération peut renvoyer les types d'erreurs suivants :
-
Erreurs renvoyées dans le cadre du code d'état HTTP :
-
AuthenticationError
-
La demande a échoué en raison d'informations d'authentification non valides.
Code d'état HTTP : 401
-
AuthorizationError
-
L'appelant n'est pas autorisé à recevoir le message.
Code d'état HTTP : 403
-
ResourceNotFoundError
-
L'invite spécifiée n'existe pas.
Code d'état HTTP : 404
-
ValidationError
-
Les arguments fournis ne répondent pas aux arguments obligatoires de l'invite.
Code d'état HTTP : 400
-
InternalServerError
-
Une erreur interne du serveur s’est produite.
Code d'état HTTP : 500
-
Erreurs MCP. Pour plus d'informations sur ces types d'erreurs, consultez la section Prompts de la documentation du Model Context Protocol (MCP).
