camel.runtime package

Subpackages

• camel.runtime.utils package
  • Submodules
  • camel.runtime.utils.function_risk_toolkit module
    • FunctionRiskToolkit
      • FunctionRiskToolkit.function_risk()
      • FunctionRiskToolkit.get_tools()
  • camel.runtime.utils.ignore_risk_toolkit module
    • IgnoreRiskToolkit
      • IgnoreRiskToolkit.add()
      • IgnoreRiskToolkit.get_tools()
      • IgnoreRiskToolkit.ignore_risk()
  • Module contents
    • FunctionRiskToolkit
      • FunctionRiskToolkit.function_risk()
      • FunctionRiskToolkit.get_tools()
    • IgnoreRiskToolkit
      • IgnoreRiskToolkit.add()
      • IgnoreRiskToolkit.get_tools()
      • IgnoreRiskToolkit.ignore_risk()

Submodules

camel.runtime.api module

camel.runtime.base module

class camel.runtime.base.BaseRuntime

Bases: ABC

An abstract base class for all CAMEL runtimes.

abstract add(funcs: FunctionTool | List[FunctionTool], *args: Any, **kwargs: Any) → BaseRuntime

Adds a new tool to the runtime.

get_tools() → List[FunctionTool]

Returns a list of all tools in the runtime.

abstract reset(*args: Any, **kwargs: Any) → Any

Resets the runtime to its initial state.

camel.runtime.configs module

class camel.runtime.configs.TaskConfig(*, cmd: str | List[str], stdout: bool = True, stderr: bool = True, stdin: bool = False, tty: bool = False, privileged: bool = False, user: str = '', detach: bool = False, stream: bool = False, socket: bool = False, environment: Dict[str, str] | List[str] | None = None, workdir: str | None = None, demux: bool = False)

Bases: BaseModel

A configuration for a task to run a command inside the container.

Arttributes:

cmd (str or list): Command to be executed stdout (bool): Attach to stdout. (default: :obj: True) stderr (bool): Attach to stderr. (default: :obj: True) stdin (bool): Attach to stdin. (default: :obj: False) tty (bool): Allocate a pseudo-TTY. (default: :obj: False) privileged (bool): Run as privileged. (default: :obj: False) user (str): User to execute command as. (default: :obj: “”) detach (bool): If true, detach from the exec command.

(default: :obj: False)

stream (bool): Stream response data. (default: :obj: False) socket (bool): Return the connection socket to allow custom

read/write operations. (default: :obj: False)

environment (dict or list): A dictionary or a list of strings in

the following format ["PASSWORD=xxx"] or {"PASSWORD": "xxx"}. (default: :obj: None)

workdir (str): Path to working directory for this exec session.

(default: :obj: None)

demux (bool): Return stdout and stderr separately. (default: :obj:

False)

cmd: str | List[str]

demux: bool

detach: bool

environment: Dict[str, str] | List[str] | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cmd': FieldInfo(annotation=Union[str, List[str]], required=True), 'demux': FieldInfo(annotation=bool, required=False, default=False), 'detach': FieldInfo(annotation=bool, required=False, default=False), 'environment': FieldInfo(annotation=Union[Dict[str, str], List[str], NoneType], required=False, default=None), 'privileged': FieldInfo(annotation=bool, required=False, default=False), 'socket': FieldInfo(annotation=bool, required=False, default=False), 'stderr': FieldInfo(annotation=bool, required=False, default=True), 'stdin': FieldInfo(annotation=bool, required=False, default=False), 'stdout': FieldInfo(annotation=bool, required=False, default=True), 'stream': FieldInfo(annotation=bool, required=False, default=False), 'tty': FieldInfo(annotation=bool, required=False, default=False), 'user': FieldInfo(annotation=str, required=False, default=''), 'workdir': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

privileged: bool

socket: bool

stderr: bool

stdin: bool

stdout: bool

stream: bool

tty: bool

user: str

workdir: str | None

camel.runtime.docker_runtime module

class camel.runtime.docker_runtime.DockerRuntime(image: str, port: int = 8000, remove: bool = True, **kwargs)

Bases: BaseRuntime

A class representing a runtime environment using Docker. This class automatically wraps functions to be executed in a Docker container.

Parameters:

• image (str) – The name of the Docker image to use for the runtime.

• port (int) – The port number to use for the runtime API. (default: :obj: 8000)

• remove (bool) – Whether to remove the container after stopping it. ‘ (default: :obj: True)

• kwargs (dict) – Additional keyword arguments to pass to the Docker client.

add(funcs: FunctionTool | List[FunctionTool], entrypoint: str, redirect_stdout: bool = False, arguments: Dict[str, Any] | None = None) → DockerRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (Union[FunctionTool, List[FunctionTool]]) – The function or list of functions to add.

• entrypoint (str) – The entrypoint for the function.

• redirect_stdout (bool) – Whether to return the stdout of the function. (default: :obj: False)

• arguments (Optional[Dict[str, Any]]) – The arguments for the function. (default: :obj: None)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

add_task(task: TaskConfig) → DockerRuntime

Add a task to run a command inside the container when building. Similar to docker exec.

Parameters:

task (TaskConfig) – The configuration for the task.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

build(time_out: int = 15) → DockerRuntime

Build the Docker container and start it.

Parameters:

time_out (int) – The number of seconds to wait for the container to start. (default: :obj: 15)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

copy(source: str, dest: str) → DockerRuntime

Copy a file or directory to the container.

Parameters:

• source (str) – The local path to the file.

• dest (str) – The path to copy the file to in the container.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

property docs: str

Get the URL for the API documentation.

Returns:

The URL for the API documentation.

Return type:

str

exec_run(task: TaskConfig) → Any

Run a command inside this container. Similar to docker exec.

Parameters:

task (TaskConfig) – The configuration for the task.

Returns:

A tuple of (exit_code, output)

exit_code: (int):

Exit code for the executed command or None if either stream or socket is True.

output: (generator, bytes, or tuple):

If stream=True, a generator yielding response chunks. If socket=True, a socket object for the connection. If demux=True, a tuple of two bytes: stdout and stderr. A bytestring containing response data otherwise.

Return type:

(ExecResult)

Raises:

RuntimeError – If the container does not exist.

mount(path: str, mount_path: str) → DockerRuntime

Mount a local directory to the container.

Parameters:

• path (str) – The local path to mount.

• mount_path (str) – The path to mount the local directory to in the container.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

property ok: bool

Check if the API Server is running.

Returns:

Whether the API Server is running.

Return type:

bool

reset() → DockerRuntime

Reset the DockerRuntime instance.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

stop(remove: bool | None = None) → DockerRuntime

stop the Docker container.

Parameters:

remove (Optional[bool]) – Whether to remove the container after stopping it. (default: :obj: None)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

wait(timeout: int = 10) → bool

Wait for the API Server to be ready.

Parameters:

timeout (int) – The number of seconds to wait. (default: :obj: 10)

Returns:

Whether the API Server is ready.

Return type:

bool

camel.runtime.llm_guard_runtime module

class camel.runtime.llm_guard_runtime.LLMGuardRuntime(prompt: str = "You are a function safety evaluator tasked with assessing the \npotential risk level of a given function based on both its description \nand parameters. Your goal is to determine if the function may pose any \nharm to the user's environment, such as deleting files, executing \narbitrary code, or accessing sensitive information. Pay special attention \nto the provided parameters — even if a function has the potential to be \ndangerous, its actual parameters may indicate harmless behavior, and the \nrisk level should be adjusted accordingly. Use the `function_risk` tool to \nassign a risk score to the function, based on the following criteria:\n\n- **Score 1**: No harm. This includes simple operations like mathematical \n    calculations, content searches, or data retrievals that do not impact \n    the user's environment or access sensitive resources. This also \n    includes functions with potentially dangerous capabilities that have \n    harmless, controlled parameters that ensure safe execution.\n- **Score 2**: Minimal harm. The function might read user files, retrieve \n    non-sensitive data, or access other low-risk resources, \n    posing little risk to the user.\n- **Score 3**: Risk present. The function might delete files, modify the \n    file system, execute arbitrary code, or access sensitive data, which \n    could negatively impact the user's environment. However, if the \n    actual parameters provided clearly indicate safe and restricted \n    usage, this risk should be downgraded accordingly.\n\nWhen evaluating, always consider both the function's description and its \nspecific parameters. If the function appears risky due to its design but \nthe provided parameters indicate a safe and non-impactful operation, \nadjust the risk score to reflect this. Assign an appropriate risk score \nand provide a brief explanation of your reasoning based on the function's \ndescription and the actual parameters given.\nYOU MUST USE THE `function_risk` TOOL TO ASSESS THE RISK \nLEVEL OF EACH FUNCTION.\n", model: BaseModelBackend | None = None, verbose: bool = False)

Bases: BaseRuntime

A runtime that evaluates the risk level of functions using a language model.

Parameters:

• prompt (str) – The prompt to use for the language model. (default: GUARDPROMPT)

• model (BaseModelBackend) – The language model to use. (default: :obj: None)

• verbose (bool) – Whether to print verbose output. (default: :obj: False)

add(funcs: FunctionTool | List[FunctionTool], threshold: int = 2) → LLMGuardRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (FunctionTool or List[FunctionTool]) – The function or list of functions to add.

• threshold (int) – The risk threshold for functions. (default: 2)

Returns:

The current runtime.

Return type:

LLMGuardRuntime

reset() → LLMGuardRuntime

Resets the runtime to its initial state.

camel.runtime.remote_http_runtime module

class camel.runtime.remote_http_runtime.RemoteHttpRuntime(host: str, port: int = 8000, python_exec: str = 'python3')

Bases: BaseRuntime

A runtime that runs functions in a remote HTTP server. You need to run the API server in the remote server first.

Parameters:

• host (str) – The host of the remote server.

• port (int) – The port of the remote server. (default: :obj: 8000)

• python_exec (str) – The python executable to run the API server. (default: :obj: python3)

add(funcs: FunctionTool | List[FunctionTool], entrypoint: str, redirect_stdout: bool = False, arguments: Dict[str, Any] | None = None) → RemoteHttpRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (Union[FunctionTool, List[FunctionTool]]) – The function or list of functions to add.

• entrypoint (str) – The entrypoint for the function.

• redirect_stdout (bool) – Whether to return the stdout of the function. (default: :obj: False)

• arguments (Optional[Dict[str, Any]]) – The arguments for the function. (default: :obj: None)

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

build() → RemoteHttpRuntime

Build the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

property docs: str

Get the URL for the API documentation.

Returns:

The URL for the API documentation.

Return type:

str

property ok: bool

Check if the API Server is running.

Returns:

Whether the API Server is running.

Return type:

bool

reset() → RemoteHttpRuntime

Reset the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

stop() → RemoteHttpRuntime

Stop the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

wait(timeout: int = 10) → bool

Wait for the API Server to be ready.

Parameters:

timeout (int) – The number of seconds to wait. (default: :obj: 10)

Returns:

Whether the API Server is ready.

Return type:

bool

Module contents

class camel.runtime.BaseRuntime

Bases: ABC

An abstract base class for all CAMEL runtimes.

abstract add(funcs: FunctionTool | List[FunctionTool], *args: Any, **kwargs: Any) → BaseRuntime

Adds a new tool to the runtime.

get_tools() → List[FunctionTool]

Returns a list of all tools in the runtime.

abstract reset(*args: Any, **kwargs: Any) → Any

Resets the runtime to its initial state.

class camel.runtime.DockerRuntime(image: str, port: int = 8000, remove: bool = True, **kwargs)

Bases: BaseRuntime

A class representing a runtime environment using Docker. This class automatically wraps functions to be executed in a Docker container.

Parameters:

• image (str) – The name of the Docker image to use for the runtime.

• port (int) – The port number to use for the runtime API. (default: :obj: 8000)

• remove (bool) – Whether to remove the container after stopping it. ‘ (default: :obj: True)

• kwargs (dict) – Additional keyword arguments to pass to the Docker client.

add(funcs: FunctionTool | List[FunctionTool], entrypoint: str, redirect_stdout: bool = False, arguments: Dict[str, Any] | None = None) → DockerRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (Union[FunctionTool, List[FunctionTool]]) – The function or list of functions to add.

• entrypoint (str) – The entrypoint for the function.

• redirect_stdout (bool) – Whether to return the stdout of the function. (default: :obj: False)

• arguments (Optional[Dict[str, Any]]) – The arguments for the function. (default: :obj: None)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

add_task(task: TaskConfig) → DockerRuntime

Add a task to run a command inside the container when building. Similar to docker exec.

Parameters:

task (TaskConfig) – The configuration for the task.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

build(time_out: int = 15) → DockerRuntime

Build the Docker container and start it.

Parameters:

time_out (int) – The number of seconds to wait for the container to start. (default: :obj: 15)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

copy(source: str, dest: str) → DockerRuntime

Copy a file or directory to the container.

Parameters:

• source (str) – The local path to the file.

• dest (str) – The path to copy the file to in the container.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

property docs: str

Get the URL for the API documentation.

Returns:

The URL for the API documentation.

Return type:

str

exec_run(task: TaskConfig) → Any

Run a command inside this container. Similar to docker exec.

Parameters:

task (TaskConfig) – The configuration for the task.

Returns:

A tuple of (exit_code, output)

exit_code: (int):

Exit code for the executed command or None if either stream or socket is True.

output: (generator, bytes, or tuple):

If stream=True, a generator yielding response chunks. If socket=True, a socket object for the connection. If demux=True, a tuple of two bytes: stdout and stderr. A bytestring containing response data otherwise.

Return type:

(ExecResult)

Raises:

RuntimeError – If the container does not exist.

mount(path: str, mount_path: str) → DockerRuntime

Mount a local directory to the container.

Parameters:

• path (str) – The local path to mount.

• mount_path (str) – The path to mount the local directory to in the container.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

property ok: bool

Check if the API Server is running.

Returns:

Whether the API Server is running.

Return type:

bool

reset() → DockerRuntime

Reset the DockerRuntime instance.

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

stop(remove: bool | None = None) → DockerRuntime

stop the Docker container.

Parameters:

remove (Optional[bool]) – Whether to remove the container after stopping it. (default: :obj: None)

Returns:

The DockerRuntime instance.

Return type:

DockerRuntime

wait(timeout: int = 10) → bool

Wait for the API Server to be ready.

Parameters:

timeout (int) – The number of seconds to wait. (default: :obj: 10)

Returns:

Whether the API Server is ready.

Return type:

bool

class camel.runtime.LLMGuardRuntime(prompt: str = "You are a function safety evaluator tasked with assessing the \npotential risk level of a given function based on both its description \nand parameters. Your goal is to determine if the function may pose any \nharm to the user's environment, such as deleting files, executing \narbitrary code, or accessing sensitive information. Pay special attention \nto the provided parameters — even if a function has the potential to be \ndangerous, its actual parameters may indicate harmless behavior, and the \nrisk level should be adjusted accordingly. Use the `function_risk` tool to \nassign a risk score to the function, based on the following criteria:\n\n- **Score 1**: No harm. This includes simple operations like mathematical \n    calculations, content searches, or data retrievals that do not impact \n    the user's environment or access sensitive resources. This also \n    includes functions with potentially dangerous capabilities that have \n    harmless, controlled parameters that ensure safe execution.\n- **Score 2**: Minimal harm. The function might read user files, retrieve \n    non-sensitive data, or access other low-risk resources, \n    posing little risk to the user.\n- **Score 3**: Risk present. The function might delete files, modify the \n    file system, execute arbitrary code, or access sensitive data, which \n    could negatively impact the user's environment. However, if the \n    actual parameters provided clearly indicate safe and restricted \n    usage, this risk should be downgraded accordingly.\n\nWhen evaluating, always consider both the function's description and its \nspecific parameters. If the function appears risky due to its design but \nthe provided parameters indicate a safe and non-impactful operation, \nadjust the risk score to reflect this. Assign an appropriate risk score \nand provide a brief explanation of your reasoning based on the function's \ndescription and the actual parameters given.\nYOU MUST USE THE `function_risk` TOOL TO ASSESS THE RISK \nLEVEL OF EACH FUNCTION.\n", model: BaseModelBackend | None = None, verbose: bool = False)

Bases: BaseRuntime

A runtime that evaluates the risk level of functions using a language model.

Parameters:

• prompt (str) – The prompt to use for the language model. (default: GUARDPROMPT)

• model (BaseModelBackend) – The language model to use. (default: :obj: None)

• verbose (bool) – Whether to print verbose output. (default: :obj: False)

add(funcs: FunctionTool | List[FunctionTool], threshold: int = 2) → LLMGuardRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (FunctionTool or List[FunctionTool]) – The function or list of functions to add.

• threshold (int) – The risk threshold for functions. (default: 2)

Returns:

The current runtime.

Return type:

LLMGuardRuntime

reset() → LLMGuardRuntime

Resets the runtime to its initial state.

class camel.runtime.RemoteHttpRuntime(host: str, port: int = 8000, python_exec: str = 'python3')

Bases: BaseRuntime

A runtime that runs functions in a remote HTTP server. You need to run the API server in the remote server first.

Parameters:

• host (str) – The host of the remote server.

• port (int) – The port of the remote server. (default: :obj: 8000)

• python_exec (str) – The python executable to run the API server. (default: :obj: python3)

add(funcs: FunctionTool | List[FunctionTool], entrypoint: str, redirect_stdout: bool = False, arguments: Dict[str, Any] | None = None) → RemoteHttpRuntime

Add a function or list of functions to the runtime.

Parameters:

• funcs (Union[FunctionTool, List[FunctionTool]]) – The function or list of functions to add.

• entrypoint (str) – The entrypoint for the function.

• redirect_stdout (bool) – Whether to return the stdout of the function. (default: :obj: False)

• arguments (Optional[Dict[str, Any]]) – The arguments for the function. (default: :obj: None)

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

build() → RemoteHttpRuntime

Build the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

property docs: str

Get the URL for the API documentation.

Returns:

The URL for the API documentation.

Return type:

str

property ok: bool

Check if the API Server is running.

Returns:

Whether the API Server is running.

Return type:

bool

reset() → RemoteHttpRuntime

Reset the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

stop() → RemoteHttpRuntime

Stop the API server.

Returns:

The current runtime.

Return type:

RemoteHttpRuntime

wait(timeout: int = 10) → bool

Wait for the API Server to be ready.

Parameters:

timeout (int) – The number of seconds to wait. (default: :obj: 10)

Returns:

Whether the API Server is ready.

Return type:

bool

class camel.runtime.TaskConfig(*, cmd: str | List[str], stdout: bool = True, stderr: bool = True, stdin: bool = False, tty: bool = False, privileged: bool = False, user: str = '', detach: bool = False, stream: bool = False, socket: bool = False, environment: Dict[str, str] | List[str] | None = None, workdir: str | None = None, demux: bool = False)

Bases: BaseModel

A configuration for a task to run a command inside the container.

Arttributes:

cmd (str or list): Command to be executed stdout (bool): Attach to stdout. (default: :obj: True) stderr (bool): Attach to stderr. (default: :obj: True) stdin (bool): Attach to stdin. (default: :obj: False) tty (bool): Allocate a pseudo-TTY. (default: :obj: False) privileged (bool): Run as privileged. (default: :obj: False) user (str): User to execute command as. (default: :obj: “”) detach (bool): If true, detach from the exec command.

(default: :obj: False)

stream (bool): Stream response data. (default: :obj: False) socket (bool): Return the connection socket to allow custom

read/write operations. (default: :obj: False)

environment (dict or list): A dictionary or a list of strings in

the following format ["PASSWORD=xxx"] or {"PASSWORD": "xxx"}. (default: :obj: None)

workdir (str): Path to working directory for this exec session.

(default: :obj: None)

demux (bool): Return stdout and stderr separately. (default: :obj:

False)

cmd: str | List[str]

demux: bool

detach: bool

environment: Dict[str, str] | List[str] | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cmd': FieldInfo(annotation=Union[str, List[str]], required=True), 'demux': FieldInfo(annotation=bool, required=False, default=False), 'detach': FieldInfo(annotation=bool, required=False, default=False), 'environment': FieldInfo(annotation=Union[Dict[str, str], List[str], NoneType], required=False, default=None), 'privileged': FieldInfo(annotation=bool, required=False, default=False), 'socket': FieldInfo(annotation=bool, required=False, default=False), 'stderr': FieldInfo(annotation=bool, required=False, default=True), 'stdin': FieldInfo(annotation=bool, required=False, default=False), 'stdout': FieldInfo(annotation=bool, required=False, default=True), 'stream': FieldInfo(annotation=bool, required=False, default=False), 'tty': FieldInfo(annotation=bool, required=False, default=False), 'user': FieldInfo(annotation=str, required=False, default=''), 'workdir': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

privileged: bool

socket: bool

stderr: bool

stdin: bool

stdout: bool

stream: bool

tty: bool

user: str

workdir: str | None