diff --git a/agentclaw/app/agent_claw_main/tools/code_interpreter.py b/agentclaw/app/agent_claw_main/tools/code_interpreter.py index f717b17..27fe973 100644 --- a/agentclaw/app/agent_claw_main/tools/code_interpreter.py +++ b/agentclaw/app/agent_claw_main/tools/code_interpreter.py @@ -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"") - 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"") - - 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'}]})