import boto3
import requests
TRUSTGATE_HOST = "https://<your-trustgate-host>"
TRUSTGATE_KEY = "<trustgate-execution-api-key>"
PROMPT_POLICY_ID = "<policy-id-for-input-checks>"
RESPONSE_POLICY_ID = "<policy-id-for-output-checks>"
bedrock = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
def actions_check(policy_id: str, text: str, user_id: str) -> dict:
resp = requests.post(
f"{TRUSTGATE_HOST}/v1/actions",
headers={"Authorization": f"Bearer {TRUSTGATE_KEY}"},
json={
"policy_id": policy_id,
"payload": {"input_text": text, "user_id": user_id},
},
timeout=10,
)
resp.raise_for_status()
return resp.json()
def ask_agent(user_id: str, prompt: str, agent_id: str, alias_id: str, session_id: str) -> str:
pre = actions_check(PROMPT_POLICY_ID, prompt, user_id)
if pre.get("status") != 200:
raise RuntimeError(f"Input blocked by TrustGate: {pre}")
sanitized_prompt = pre["payload"].get("result", prompt)
response = bedrock.invoke_agent(
agentId=agent_id,
agentAliasId=alias_id,
sessionId=session_id,
inputText=sanitized_prompt,
)
completion = "".join(
event["chunk"]["bytes"].decode()
for event in response.get("completion", [])
if "chunk" in event
)
post = actions_check(RESPONSE_POLICY_ID, completion, user_id)
if post.get("status") != 200:
raise RuntimeError(f"Output blocked by TrustGate: {post}")
return post["payload"].get("result", completion)