fix: align run_code with AWS docs pattern (invoke+executeCode, not execute_code wrapper)
This commit is contained in:
@@ -1,68 +1,42 @@
|
||||
"""Code interpreter tool — runs Python code in AgentCore managed sandbox."""
|
||||
"""Code interpreter tool — runs Python code in AgentCore managed sandbox.
|
||||
|
||||
Follows the AWS-recommended pattern:
|
||||
https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/code-interpreter-building-agents.html
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import base64
|
||||
from strands import tool
|
||||
|
||||
|
||||
def _parse_stream(result: dict) -> str:
|
||||
"""Parse the streaming response from invoke_code_interpreter."""
|
||||
parts = []
|
||||
if "stream" not in result:
|
||||
return str(result)
|
||||
|
||||
for event in result["stream"]:
|
||||
if "result" not in event:
|
||||
continue
|
||||
for item in event["result"].get("content", []):
|
||||
item_type = item.get("type", "")
|
||||
if item_type == "text":
|
||||
text = item.get("text", "")
|
||||
if text:
|
||||
parts.append(text)
|
||||
elif item_type == "resource":
|
||||
resource = item.get("resource", {})
|
||||
if "text" in resource:
|
||||
parts.append(resource["text"])
|
||||
elif "blob" in resource:
|
||||
try:
|
||||
parts.append(base64.b64decode(resource["blob"]).decode("utf-8"))
|
||||
except Exception:
|
||||
parts.append(f"<binary resource: {resource.get('uri', '?')}>")
|
||||
elif item_type == "image":
|
||||
# Base64-encoded image
|
||||
image_data = item.get("source", {}).get("data", "")
|
||||
mime = item.get("source", {}).get("mediaType", "image/png")
|
||||
parts.append(f"<image: {mime}, {len(image_data)} bytes base64>")
|
||||
|
||||
return "\n".join(parts) if parts else "(no output)"
|
||||
from bedrock_agentcore.tools.code_interpreter_client import code_session
|
||||
|
||||
|
||||
@tool
|
||||
def run_code(code: str, packages: list[str] | None = None) -> str:
|
||||
"""Execute Python code in a secure managed sandbox and return the output.
|
||||
Optionally install pip packages before running (e.g. ['pandas', 'numpy']).
|
||||
def run_code(code: str, description: str = '') -> str:
|
||||
"""Execute Python code in a secure AgentCore managed sandbox and return the output.
|
||||
State is maintained within a single session but not across separate calls.
|
||||
Supports data analysis, calculations, file I/O, and most common Python libraries.
|
||||
|
||||
Args:
|
||||
code: Python code to execute.
|
||||
packages: Optional list of pip packages to install first.
|
||||
description: Optional description prepended as a comment.
|
||||
|
||||
Returns:
|
||||
Execution output (stdout, results, errors).
|
||||
JSON result with stdout, stderr, exitCode, and executionTime.
|
||||
"""
|
||||
try:
|
||||
from bedrock_agentcore.tools import CodeInterpreter, code_session
|
||||
if description:
|
||||
code = f'# {description}\n{code}'
|
||||
|
||||
region = os.environ.get('AWS_REGION', 'us-east-1')
|
||||
print(f'[run_code] executing {len(code)}c')
|
||||
|
||||
with code_session(region) as client:
|
||||
if packages:
|
||||
install_raw = client.install_packages(packages)
|
||||
install_out = _parse_stream(install_raw) if isinstance(install_raw, dict) else str(install_raw)
|
||||
print(f'[code_interpreter] install: {install_out[:200]}')
|
||||
region = os.environ.get('AWS_REGION', 'us-east-1')
|
||||
|
||||
raw = client.execute_code(code)
|
||||
return _parse_stream(raw)
|
||||
with code_session(region) as client:
|
||||
response = client.invoke('executeCode', {
|
||||
'code': code,
|
||||
'language': 'python',
|
||||
'clearContext': False,
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
return f'Code interpreter error: {type(e).__name__}: {e}\n{traceback.format_exc()[-500:]}'
|
||||
for event in response['stream']:
|
||||
return json.dumps(event['result'])
|
||||
|
||||
return json.dumps({'isError': True, 'content': [{'type': 'text', 'text': 'No output from code interpreter'}]})
|
||||
|
||||
Reference in New Issue
Block a user