Skip to content

Conversation

@sundargthb
Copy link
Contributor

Summary

Adds developer-friendly convenience methods to CodeInterpreter class, reducing boilerplate for common operations while maintaining full backward compatibility.

Motivation

Current usage requires verbose invoke() calls with manual parameter construction:

# Current - verbose
client.invoke('writeFiles', {'content': [{'path': 'data.csv', 'text': content}]})
client.invoke('executeCommand', {'command': 'pip install pandas numpy'})

This PR enables cleaner patterns that match developer expectations:

# New - intuitive
client.upload_file('data.csv', content, description='Sales data')
client.install_packages(['pandas', 'numpy'])

Changes

New Methods

Method Purpose
upload_file(path, content, description) Upload file with semantic context
upload_files(files) Batch file upload
install_packages(packages, upgrade) Install pip packages safely
download_file(path) Read file content
download_files(paths) Batch file read
execute_code(code, language, clear_context) Typed code execution
execute_shell(command) Shell command wrapper

Security

  • Rejects absolute paths (/etc/passwdValueError)
  • Blocks shell injection in package names (;, &, |, `, $)
  • Validates language parameter

Backward Compatibility

No breaking changes - All existing invoke() patterns continue to work unchanged.

Testing

  • Unit tests with mocked invoke() calls
  • Integration tests against live AgentCore environment
  • Session reuse verification

@sundargthb sundargthb deployed to auto-approve January 1, 2026 00:26 — with GitHub Actions Active
result = self.invoke("writeFiles", {"content": [file_content]})

# Store description as metadata (available for future LLM context)
if description and hasattr(self, "_file_descriptions"):

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need the hasattr check here? I see _file_descriptions is defined above as {}?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will remove this..

file_contents.append({"path": path, "text": content})

self.logger.info("Uploading %d files", len(files))
return self.invoke("writeFiles", {"content": file_contents})

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

didn't follow this. we are taking list of files and there is no batch file upload API in code interpreter? so we should just call upload_file utility we have above?

Also need to see how to handle partial failure in this case. Throwing an exception even if one file upload files should be ok and users will have to retry entire batch. Some advanced support would be to instead return list of responses back and let clients handle which ones to retry with

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The writeFiles API does accept a list, so it is a batch operation at the API level. But your point about reusing upload_file and handling partial failures is valid.

Two options:

  • Keep as-is since writeFiles API handles the batch natively (simpler, atomic)
  • Loop through and call upload_file for each, collect results, and let caller handle partial failures

I'd lean toward Option A for now since the API handles it atomically - either all succeed or all fail. We can add advanced partial-failure handling as a follow-up if users request it.

path: str,
content: Union[str, bytes],
description: str = "",
) -> Dict[str, Any]:

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

are we not able to strengthen the return type further here? what's the expected shape we plan to return here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The return type is Dict[str, Any] because invoke() returns that. We could define a TypedDict for the response shape, but it couples us to the API response structure. For now, keeping Dict[str, Any] gives flexibility. Can revisit if we want stricter typing across the SDK.

self.logger.info("Downloading %d files", len(paths))
result = self.invoke("readFiles", {"paths": paths})

files = {}

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we re_use the utility method above to download_file?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, makes sense. Will refactor to reuse download_file

self,
code: str,
language: str = "python",
clear_context: bool = False,
Copy link

@aws-bviswa aws-bviswa Jan 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor note: this clear_context is only application for python and is not supported for other languages yet.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. Will update the docstring to clarify this limitation.

code: The code to execute.
language: Programming language - 'python', 'javascript', or 'typescript'.
Default is 'python'.
clear_context: If True, clears all previous variable state before execution.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we consider adding a first class method to clear context? so that users can write code such as:

client.executeCode(`x = 10`)
client.executeCode(`x += 1`)
client.clearContext() // this can be a dummy call we run with clearContext = True
client.executeCode(`x = 1`)

so it's easy and clean to clear context instead of always needing to set it along with another execute code call.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cleaner API. Will add.

},
)

def execute_shell(

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we call this exec() or execute_command() to keep consistency with API?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure, execute_command matches the API name better. Will rename.

self.logger.info("Installing packages: %s", packages_str)
return self.invoke("executeCommand", {"command": command})

def download_file(

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we call this read/write file to keep it mapped to API input names? (we may introduce download/upload for some future operations related to blob storage upload/download)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think upload_file/download_file are more intuitive from a user perspective than write_file/read_file. The existing invoke('writeFiles') and invoke('readFiles') are the low-level API - these convenience methods add some semantic clarity.

Suggest we keep upload_file/download_file for now, and if we add blob storage later, we can name those upload_to_s3/download_from_s3 or similar.

identifier (str, optional): The code interpreter identifier.
session_id (str, optional): The active session ID.
Basic Usage:

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what if we could have subclasses to organize the utility methods based on:

  • file system controls
  • code execution controls
  • command line controls

so we would end up with:

session.file.read()
session.file.write()

session.code.execute()

session.cmd.execute()
session.cmd.startTask()
session.cmd.getTask()
session.cmd.stopTask()

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we keep this as a follow-up enhancement. Current flat structure works and matches how other SDKs typically expose methods. We can layer subclasses on top in a future version without breaking changes.

self,
command: str,
) -> Dict[str, Any]:
"""Execute a shell command in the interpreter environment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we also have few command task related API's we could have wrappers for

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/code-interpreter-api-reference-examples.html

this can also be a follow-up

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree these would be useful. Already have invoke() paths for these. Can add as follow-up since the core file/code operations are the priority.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants