From 74bad6b16198536268bf0ffb240acf87095a85ee Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 8 Sep 2025 16:46:35 +0000 Subject: [PATCH] Add AWS Bedrock examples for Converse API and Managed Agents Co-authored-by: sgrantbabb --- README_BEDROCK.md | 57 +++++++++++++ bedrock_converse_agent.py | 166 ++++++++++++++++++++++++++++++++++++++ bedrock_managed_agents.py | 73 +++++++++++++++++ requirements.txt | 6 ++ 4 files changed, 302 insertions(+) create mode 100644 README_BEDROCK.md create mode 100644 bedrock_converse_agent.py create mode 100644 bedrock_managed_agents.py diff --git a/README_BEDROCK.md b/README_BEDROCK.md new file mode 100644 index 00000000..8e743165 --- /dev/null +++ b/README_BEDROCK.md @@ -0,0 +1,57 @@ +# AWS Bedrock Agent Examples + +Two approaches are included: + +- Converse API lightweight agent with tool use: `bedrock_converse_agent.py` +- Managed Agents for Bedrock scripts: `bedrock_managed_agents.py` + +## Setup + +1) Python deps + +```bash +python -m venv .venv && source .venv/bin/activate +pip install -r requirements.txt +``` + +2) AWS credentials + +- Configure credentials with permissions for `bedrock` and `bedrock-runtime`. +- Set `AWS_REGION` or `AWS_DEFAULT_REGION`. +- Optionally set `BEDROCK_MODEL_ID` to override the default model. + +## Converse API agent + +Run a prompt through a small agent loop with a demo tool `get_weather`. + +```bash +python bedrock_converse_agent.py --prompt "What's the weather in London in F?" +``` + +Options: + +- `--model-id`: override model (default: `anthropic.claude-3-5-sonnet-20240620-v1:0`) +- `--region`: AWS region override + +## Managed Agents for Bedrock + +Create, prepare, and invoke a managed Agent. + +```bash +# Create agent +python bedrock_managed_agents.py create-agent \ + --name demo-agent \ + --foundation-model anthropic.claude-3-5-sonnet-20240620-v1:0 \ + --instruction "You are a helpful assistant." + +# Prepare (deploy) agent +python bedrock_managed_agents.py prepare-agent --agent-id + +# Invoke agent +python bedrock_managed_agents.py invoke-agent --agent-id --input-text "Hello" +``` + +Notes: + +- Ensure your account has access to the chosen foundation model. +- For production, add tools/knowledge bases/guardrails to the managed agent as needed. \ No newline at end of file diff --git a/bedrock_converse_agent.py b/bedrock_converse_agent.py new file mode 100644 index 00000000..046ac147 --- /dev/null +++ b/bedrock_converse_agent.py @@ -0,0 +1,166 @@ +import json +import os +from typing import Any, Dict, List, Callable, Optional, Tuple + +import boto3 +from botocore.config import Config +import click +from rich.console import Console +from rich.panel import Panel + + +console = Console() + + +def get_bedrock_runtime_client(region_name: Optional[str] = None): + region = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") + if not region: + raise RuntimeError("AWS region not set. Provide --region or set AWS_REGION/AWS_DEFAULT_REGION.") + return boto3.client("bedrock-runtime", region_name=region, config=Config(retries={"max_attempts": 10})) + + +def get_default_model_id() -> str: + # Popular high-quality model on Bedrock; change if your account lacks access + return os.getenv("BEDROCK_MODEL_ID", "anthropic.claude-3-5-sonnet-20240620-v1:0") + + +class Tool: + def __init__(self, name: str, description: str, json_schema: Dict[str, Any], impl: Callable[[Dict[str, Any]], Dict[str, Any]]): + self.name = name + self.description = description + self.json_schema = json_schema + self.impl = impl + + def to_tool_spec(self) -> Dict[str, Any]: + return { + "toolSpec": { + "name": self.name, + "description": self.description, + "inputSchema": {"json": self.json_schema}, + } + } + + +def get_weather_impl(args: Dict[str, Any]) -> Dict[str, Any]: + city = str(args.get("city", "")) + unit = str(args.get("unit", "c")).lower() + if unit not in {"c", "f"}: + unit = "c" + # Dummy data; in real use, call a weather API here + base_temp_c = 22.3 + if unit == "f": + temp = base_temp_c * 9 / 5 + 32 + else: + temp = base_temp_c + return {"city": city, "unit": unit, "temperature": round(temp, 1), "conditions": "Sunny"} + + +def build_tool_registry() -> Dict[str, Tool]: + return { + "get_weather": Tool( + name="get_weather", + description="Get current weather for a city.", + json_schema={ + "type": "object", + "properties": { + "city": {"type": "string", "description": "City name, e.g., London"}, + "unit": {"type": "string", "enum": ["c", "f"], "default": "c"}, + }, + "required": ["city"], + }, + impl=get_weather_impl, + ), + } + + +def extract_tool_uses_from_content(content: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + tool_uses: List[Dict[str, Any]] = [] + for block in content: + if "toolUse" in block: + tool_uses.append(block["toolUse"]) + return tool_uses + + +def build_tool_results(tool_uses: List[Dict[str, Any]], registry: Dict[str, Tool]) -> List[Dict[str, Any]]: + results: List[Dict[str, Any]] = [] + for tu in tool_uses: + name = tu.get("name") + tool_use_id = tu.get("toolUseId") + inputs = tu.get("input") or {} + impl = registry.get(name) + if not impl: + results.append({ + "toolResult": { + "toolUseId": tool_use_id, + "content": [{"text": f"Tool '{name}' not implemented."}], + "status": "error", + } + }) + continue + try: + output_json = impl.impl(inputs) + results.append({ + "toolResult": { + "toolUseId": tool_use_id, + "content": [{"json": output_json}], + "status": "success", + } + }) + except Exception as exc: # pragma: no cover (defensive) + results.append({ + "toolResult": { + "toolUseId": tool_use_id, + "content": [{"text": f"Tool '{name}' failed: {exc}"}], + "status": "error", + } + }) + return results + + +def run_converse_agent(prompt: str, model_id: Optional[str] = None, region: Optional[str] = None, system_prompt: Optional[str] = None) -> Tuple[str, List[Dict[str, Any]]]: + client = get_bedrock_runtime_client(region) + model = model_id or get_default_model_id() + tools = build_tool_registry() + tool_config = {"tools": [t.to_tool_spec() for t in tools.values()]} + system_blocks = [{"text": system_prompt or "You are a concise, helpful agent. Use tools when beneficial."}] + messages: List[Dict[str, Any]] = [{"role": "user", "content": [{"text": prompt}]}] + + final_text: str = "" + for _ in range(5): + resp = client.converse(modelId=model, system=system_blocks, toolConfig=tool_config, messages=messages) + assistant_msg = resp.get("output", {}).get("message", {}) + stop_reason = resp.get("stopReason") + content = assistant_msg.get("content", []) + + # Gather any assistant text so far + for block in content: + if "text" in block: + final_text += block["text"] + + tool_uses = extract_tool_uses_from_content(content) + if tool_uses: + results = build_tool_results(tool_uses, tools) + messages.append(assistant_msg) + messages.append({"role": "user", "content": results}) + continue + + # If no tool use requested, or completed after tool results + if stop_reason in {"end_turn", "max_tokens"} or not tool_uses: + break + + return final_text.strip(), messages + + +@click.command() +@click.option("--prompt", required=True, help="User prompt to send to the agent") +@click.option("--model-id", default=lambda: get_default_model_id(), show_default=True, help="Bedrock model ID") +@click.option("--region", default=lambda: os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "", help="AWS region") +def main(prompt: str, model_id: str, region: str): + """Run a lightweight agent loop on Bedrock using the Converse API.""" + final_text, _ = run_converse_agent(prompt=prompt, model_id=model_id or None, region=region or None) + console.print(Panel.fit(final_text or "(no content)", title="Assistant", subtitle=model_id)) + + +if __name__ == "__main__": + main() + diff --git a/bedrock_managed_agents.py b/bedrock_managed_agents.py new file mode 100644 index 00000000..831a0b50 --- /dev/null +++ b/bedrock_managed_agents.py @@ -0,0 +1,73 @@ +import json +import os +from typing import Any, Dict, Optional + +import boto3 +from botocore.config import Config +import click +from rich.console import Console +from rich.panel import Panel + + +console = Console() + + +def get_bedrock_clients(region_name: Optional[str] = None): + region = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") + if not region: + raise RuntimeError("AWS region not set. Provide --region or set AWS_REGION/AWS_DEFAULT_REGION.") + return ( + boto3.client("bedrock", region_name=region, config=Config(retries={"max_attempts": 10})), + boto3.client("bedrock-runtime", region_name=region, config=Config(retries={"max_attempts": 10})), + ) + + +@click.group() +def cli(): + """Scripts to create and invoke an Agent for Amazon Bedrock.""" + pass + + +@cli.command("create-agent") +@click.option("--name", required=True, help="Agent name") +@click.option("--foundation-model", required=True, help="Model ID, e.g., anthropic.claude-3-5-sonnet-20240620-v1:0") +@click.option("--instruction", required=True, help="System prompt / instruction for the agent") +@click.option("--region", default=lambda: os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "", help="AWS region") +def create_agent(name: str, foundation_model: str, instruction: str, region: str): + bedrock, _ = get_bedrock_clients(region) + resp = bedrock.create_agent( + agentName=name, + foundationModel=foundation_model, + instruction=instruction, + offGuardrail=False, + ) + agent_id = resp["agent"].get("agentId") + console.print(Panel.fit(f"Created agent: {agent_id}", title="Create Agent")) + console.print(json.dumps(resp, indent=2)) + + +@cli.command("prepare-agent") +@click.option("--agent-id", required=True, help="Agent ID from create-agent") +@click.option("--region", default=lambda: os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "", help="AWS region") +def prepare_agent(agent_id: str, region: str): + bedrock, _ = get_bedrock_clients(region) + resp = bedrock.prepare_agent(agentId=agent_id) + console.print(Panel.fit(f"Preparing agent: {agent_id}", title="Prepare Agent")) + console.print(json.dumps(resp, indent=2)) + + +@cli.command("invoke-agent") +@click.option("--agent-id", required=True, help="Agent ID") +@click.option("--input-text", required=True, help="User input") +@click.option("--session-id", default=None, help="Optional session ID for continuity") +@click.option("--region", default=lambda: os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "", help="AWS region") +def invoke_agent(agent_id: str, input_text: str, session_id: Optional[str], region: str): + bedrock, bedrock_runtime = get_bedrock_clients(region) + resp = bedrock_runtime.invoke_agent(agentId=agent_id, sessionId=session_id or "session-1", inputText=input_text) + completion = resp.get("completion") + console.print(Panel.fit(completion or "(no content)", title="Agent Response")) + + +if __name__ == "__main__": + cli() + diff --git a/requirements.txt b/requirements.txt index 9c9d910a..bdba1d57 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,9 @@ +boto3>=1.34.140 +botocore>=1.34.140 +click>=8.1.7 +rich>=13.7.1 +python-dotenv>=1.0.1 +typing-extensions>=4.12.2 torch==2.3.1 torchvision==0.18.1 diffusers==0.11.1