Engine 模块 ¶

async stop(deploy_id, **kwargs)[source]

Stop the FastAPI service.

Parameters:

deploy_id (str) – Deployment identifier
**kwargs – Additional parameters

Returns:

Dict with success status, message, and details

Return type:

is_service_running()[source]

Check if service is running.

Return type:: bool

get_deployment_info()[source]

Get deployment information.

Return type:: Dict[str, Any]

property service_url: str | None: Get the current service URL if running.

class agentscope_runtime.engine.deployers.kubernetes_deployer.K8sConfig(*, k8s_namespace='agentscope-runtime', kubeconfig_path=None)[source]

Bases: BaseModel

Parameters:

k8s_namespace (str | None)
kubeconfig_path (str | None)

k8s_namespace: str | None

kubeconfig_path: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.kubernetes_deployer.BuildConfig(*, build_context_dir=None, dockerfile_template=None, build_timeout=600, push_timeout=300, cleanup_after_build=True)[source]

Bases: BaseModel

Build configuration

Parameters:

build_context_dir (str | None)
dockerfile_template (str)
build_timeout (int)
push_timeout (int)
cleanup_after_build (bool)

build_context_dir: str | None

dockerfile_template: str

build_timeout: int

push_timeout: int

cleanup_after_build: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.kubernetes_deployer.KubernetesDeployManager(kube_config=None, registry_config=RegistryConfig(registry_url='', username=None, password=None, namespace='agentscope-runtime', image_pull_secret=None), use_deployment=True, build_context_dir=None, state_manager=None)[source]

Bases: DeployManager

Kubernetes deployer for agent services

Parameters:

kube_config (K8sConfig)
registry_config (RegistryConfig)
use_deployment (bool)
build_context_dir (str | None)

__init__(kube_config=None, registry_config=RegistryConfig(registry_url='', username=None, password=None, namespace='agentscope-runtime', image_pull_secret=None), use_deployment=True, build_context_dir=None, state_manager=None)[source]

Parameters:

kube_config (K8sConfig | None)
registry_config (RegistryConfig)
use_deployment (bool)
build_context_dir (str | None)

static get_service_endpoint(service_external_ip, service_port, fallback_host='127.0.0.1')[source]

Auto-select appropriate service endpoint based on detected environment.

Solves the common issue where Kubernetes LoadBalancer/ExternalIP is not reachable from localhost in local clusters (e.g., Minikube/Kind).

Parameters:

service_external_ip (str | None) – ExternalIP or LoadBalancer IP from Service
service_port (int | list | None) – Target port
fallback_host (str) – Host to use in local environments (default:
127.0.0.1)

Returns:

Full HTTP endpoint URL: http://<host>:<port>

Return type:

Example

>>> endpoint = get_service_endpoint('192.168.5.1', 8080)
>>> # In local env → 'http://127.0.0.1:8080'
>>> # In cloud env → 'http://192.168.5.1:8080'

static get_resource_name(deploy_id)[source]

Parameters:: deploy_id (str)
Return type:: str

async deploy(app=None, runner=None, entrypoint=None, endpoint_path='/process', stream=True, custom_endpoints=None, protocol_adapters=None, requirements=None, extra_packages=None, base_image='python:3.9-slim', environment=None, runtime_config=None, port=8090, replicas=1, mount_dir=None, image_name='agent_llm', image_tag='latest', push_to_registry=False, use_cache=True, **kwargs)[source]

Deploy runner to Kubernetes.

All temporary files are created in cwd/.agentscope_runtime/ by default.

Parameters:

app – Agent app to be deployed
runner – Complete Runner object with agent, environment_manager, context_manager
entrypoint (str | None) – Entrypoint spec (e.g., “app.py” or “app.py:handler”)
endpoint_path (str) – API endpoint path
stream (bool) – Enable streaming responses
custom_endpoints (List[Dict] | None) – Custom endpoints from agent app
protocol_adapters (list[ProtocolAdapter] | None) – protocol adapters
requirements (str | None | List[str]) – PyPI dependencies (following _agent_engines.py pattern)
extra_packages (List[str] | None) – User code directory/file path
base_image (str) – Docker base image
port (int) – Container port
replicas (int) – Number of replicas
environment (Dict | None) – Environment variables dict
mount_dir (str | None) – Mount directory
runtime_config (Dict | None) – K8s runtime configuration
use_cache (bool) – Enable build cache (default: True)
compatibility (# Backward)
image_name (str) – Image name
image_tag (str) – Image tag
push_to_registry (bool) – Push to registry
**kwargs – Additional arguments

Returns:

Dict containing deploy_id, url, resource_name, replicas

Raises:

RuntimeError – If deployment fails

Return type:

async stop(deploy_id, **kwargs)[source]

Stop Kubernetes deployment.

Parameters:

deploy_id (str) – Deployment identifier
**kwargs – Additional parameters

Returns:

Dict with success status, message, and details

Return type:

get_status()[source]

Get deployment status

Return type:: str

class agentscope_runtime.engine.deployers.modelstudio_deployer.OSSConfig(*, region='cn-hangzhou', access_key_id=None, access_key_secret=None, bucket_prefix='tmpbucket-agentscope-runtime')[source]

Bases: BaseModel

Parameters:

region (str)
access_key_id (str | None)
access_key_secret (str | None)
bucket_prefix (str)

region: str

access_key_id: str | None

access_key_secret: str | None

bucket_prefix: str

classmethod from_env()[source]

Return type:: OSSConfig

ensure_valid()[source]

Return type:: None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.modelstudio_deployer.ModelstudioConfig(*, endpoint='bailian.cn-beijing.aliyuncs.com', workspace_id=None, access_key_id=None, access_key_secret=None, dashscope_api_key=None)[source]

Bases: BaseModel

Parameters:

endpoint (str)
workspace_id (str | None)
access_key_id (str | None)
access_key_secret (str | None)
dashscope_api_key (str | None)

endpoint: str

workspace_id: str | None

access_key_id: str | None

access_key_secret: str | None

dashscope_api_key: str | None

classmethod from_env()[source]

Return type:: ModelstudioConfig

ensure_valid()[source]

Return type:: None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.modelstudio_deployer.ModelstudioDeployManager(oss_config=None, modelstudio_config=None, build_root=None, state_manager=None)[source]

Bases: DeployManager

Deployer for Alibaba Modelstudio Function Compute based agent deployment.

This deployer packages the user project into a wheel, uploads it to OSS, and triggers a Modelstudio Full-Code deploy.

Parameters:

oss_config (OSSConfig | None)
modelstudio_config (ModelstudioConfig | None)
build_root (str | Path | None)

__init__(oss_config=None, modelstudio_config=None, build_root=None, state_manager=None)[source]

Parameters:

oss_config (OSSConfig | None)
modelstudio_config (ModelstudioConfig | None)
build_root (str | Path | None)

Return type:

None

async deploy(app=None, runner=None, endpoint_path='/process', protocol_adapters=None, requirements=None, extra_packages=None, environment=None, project_dir=None, cmd=None, deploy_name=None, skip_upload=False, telemetry_enabled=True, external_whl_path=None, agent_id=None, agent_desc=None, custom_endpoints=None, **kwargs)[source]

Package the project, upload to OSS and trigger ModelStudio deploy.

Returns a dict containing deploy_id, wheel_path, url (if uploaded), resource_name (deploy_name), and workspace_id.

Parameters:

endpoint_path (str)
protocol_adapters (list[ProtocolAdapter] | None)
requirements (str | None | List[str])
extra_packages (List[str] | None)
environment (Dict[str, str] | None)
project_dir (str | Path | None)
cmd (str | None)
deploy_name (str | None)
skip_upload (bool)
telemetry_enabled (bool)
external_whl_path (str | None)
agent_id (str | None)
agent_desc (str | None)
custom_endpoints (List[Dict] | None)

Return type:

Dict[str, str]

async stop(deploy_id, **kwargs)[source]

Stop ModelStudio deployment.

Note: ModelStudio stop API not yet available.

Parameters:

deploy_id (str) – Deployment identifier
**kwargs – Additional parameters

Returns:

Dict with success status and message

Return type:

get_status()[source]

Return type:: str

class agentscope_runtime.engine.deployers.agentrun_deployer.EndpointConfig(agent_runtime_endpoint_name=None, description=None, tags=None, target_version='LATEST')[source]

Bases: object

Configuration for agent runtime endpoint.

Parameters:

agent_runtime_endpoint_name (str | None)
description (str | None)
tags (List[str] | None)
target_version (str | None)

agent_runtime_endpoint_name: str | None = None

description: str | None = None

tags: List[str] | None = None

target_version: str | None = 'LATEST'

__init__(agent_runtime_endpoint_name=None, description=None, tags=None, target_version='LATEST')

Parameters:

agent_runtime_endpoint_name (str | None)
description (str | None)
tags (List[str] | None)
target_version (str | None)

Return type:

None

class agentscope_runtime.engine.deployers.agentrun_deployer.CodeConfig(command=None, oss_bucket_name=None, oss_object_name=None)[source]

Bases: object

Configuration for code-based runtimes.

Parameters:

command (List[str] | None)
oss_bucket_name (str | None)
oss_object_name (str | None)

command: List[str] | None = None

oss_bucket_name: str | None = None

oss_object_name: str | None = None

__init__(command=None, oss_bucket_name=None, oss_object_name=None)

Parameters:

command (List[str] | None)
oss_bucket_name (str | None)
oss_object_name (str | None)

Return type:

None

class agentscope_runtime.engine.deployers.agentrun_deployer.LogConfig(logstore=None, project=None)[source]

Bases: object

Configuration for logging.

Parameters:

logstore (str | None)
project (str | None)

logstore: str | None = None

project: str | None = None

__init__(logstore=None, project=None)

Parameters:

logstore (str | None)
project (str | None)

Return type:

None

class agentscope_runtime.engine.deployers.agentrun_deployer.NetworkConfig(network_mode='PUBLIC', security_group_id=None, vpc_id=None, vswitch_ids=None)[source]

Bases: object

Network configuration for the runtime.

Parameters:

network_mode (str)
security_group_id (str | None)
vpc_id (str | None)
vswitch_ids (list[str] | None)

network_mode: str = 'PUBLIC'

security_group_id: str | None = None

vpc_id: str | None = None

vswitch_ids: list[str] | None = None

__init__(network_mode='PUBLIC', security_group_id=None, vpc_id=None, vswitch_ids=None)

Parameters:

network_mode (str)
security_group_id (str | None)
vpc_id (str | None)
vswitch_ids (list[str] | None)

Return type:

None

class agentscope_runtime.engine.deployers.agentrun_deployer.AgentRunConfig(*, access_key_id=None, access_key_secret=None, region_id='cn-hangzhou', endpoint=None, log_config=None, network_config=NetworkConfig(network_mode='PUBLIC', security_group_id=None, vpc_id=None, vswitch_ids=None), cpu=2.0, memory=2048, execution_role_arn=None, session_concurrency_limit=200, session_idle_timeout_seconds=3600)[source]

Bases: BaseModel

Parameters:

access_key_id (str | None)
access_key_secret (str | None)
region_id (str)
endpoint (str | None)
log_config (LogConfig | None)
network_config (NetworkConfig | None)
cpu (float)
memory (int)
execution_role_arn (str | None)
session_concurrency_limit (int | None)
session_idle_timeout_seconds (int | None)

access_key_id: str | None

access_key_secret: str | None

region_id: str

endpoint: str | None

log_config: LogConfig | None

network_config: NetworkConfig | None

cpu: float

memory: int

execution_role_arn: str | None

session_concurrency_limit: int | None

session_idle_timeout_seconds: int | None

classmethod from_env()[source]

Create AgentRunConfig from environment variables.

Returns:: Configuration loaded from environment variables.
Return type:: AgentRunConfig

ensure_valid()[source]

Validate that all required configuration fields are present.

Raises:: ValueError – If required environment variables are missing.
Return type:: None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.agentrun_deployer.OSSConfig(*, region='cn-hangzhou', access_key_id=None, access_key_secret=None, bucket_name)[source]

Bases: BaseModel

Parameters:

region (str)
access_key_id (str | None)
access_key_secret (str | None)
bucket_name (str)

region: str

access_key_id: str | None

access_key_secret: str | None

bucket_name: str

classmethod from_env()[source]

Create OSSConfig from environment variables.

Returns:: Configuration loaded from environment variables.
Return type:: OSSConfig

ensure_valid()[source]

Validate that all required OSS configuration fields are present.

Raises:: RuntimeError – If required AccessKey credentials are missing.
Return type:: None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.agentrun_deployer.AgentRunDeployManager(oss_config=None, agentrun_config=None, build_root=None, state_manager=None)[source]

Bases: DeployManager

Manager for deploying agents to Alibaba Cloud AgentRun service.

This class handles the complete deployment workflow including: - Building and packaging agent projects - Uploading artifacts to OSS - Creating and managing agent runtimes - Creating and managing runtime endpoints

Parameters:

oss_config (OSSConfig | None)
agentrun_config (AgentRunConfig | None)
build_root (str | Path | None)

GET_AGENT_RUNTIME_STATUS_MAX_ATTEMPTS = 60

GET_AGENT_RUNTIME_STATUS_INTERVAL = 1

LATEST_VERSION = 'LATEST'

DEFAULT_ENDPOINT_NAME = 'default-endpoint'

__init__(oss_config=None, agentrun_config=None, build_root=None, state_manager=None)[source]

Initialize AgentRun deployment manager.

Parameters:

oss_config (OSSConfig | None) – OSS configuration for artifact storage. If None, loads from environment.
agentrun_config (AgentRunConfig | None) – AgentRun service configuration. If None, loads from environment.
build_root (str | Path | None) – Root directory for build artifacts. If None, uses parent directory of current working directory.
state_manager – Deployment state manager. If None, creates a new instance.

async deploy(runner=None, endpoint_path='/process', protocol_adapters=None, requirements=None, extra_packages=None, environment=None, project_dir=None, cmd=None, deploy_name=None, skip_upload=False, external_whl_path=None, agentrun_id=None, custom_endpoints=None, app=None, **kwargs)[source]

Deploy agent to AgentRun service.

Parameters:

app – AgentApp instance to deploy.
runner – Runner instance containing the agent to deploy.
endpoint_path (str) – HTTP endpoint path for the agent service.
protocol_adapters (list[ProtocolAdapter] | None) – List of protocol adapters for the agent.
requirements (str | None | List[str]) – Python requirements for the agent (file path or list).
extra_packages (List[str] | None) – Additional Python packages to install.
environment (Dict[str, str] | None) – Environment variables for the runtime.
project_dir (str | Path | None) – Project directory to deploy.
cmd (str | None) – Command to start the agent application.
deploy_name (str | None) – Name for this deployment.
skip_upload (bool) – If True, skip uploading to AgentRun service.
external_whl_path (str | None) – Path to pre-built wheel file.
agentrun_id (str | None) – ID of existing agent to update.
custom_endpoints (List[Dict] | None) – Custom endpoints for the agent.
**kwargs – Additional deployment parameters.

Returns:

success: Whether deployment succeeded
message: Status message
agentrun_id: ID of the agent runtime
agentrun_endpoint_url: Public endpoint URL
build_dir: Build directory path
wheel_path: Path to wheel file
artifact_url: OSS artifact URL
url: Console URL for the deployment
deploy_id: Deployment ID
resource_name: Resource name

Return type:

Dictionary containing deployment results with keys

Raises:

ValueError – If required parameters are missing.
FileNotFoundError – If specified files/directories don’t exist.

async deploy_to_agentrun(agent_runtime_name, oss_bucket_name, oss_object_name, agentrun_id=None, environment=None)[source]

Deploy agent runtime and endpoint to AgentRun service.

Parameters:

agent_runtime_name (str) – Name for the agent runtime.
oss_bucket_name (str) – OSS bucket containing the code artifact.
oss_object_name (str) – Object key of the code artifact in OSS.
agentrun_id (str | None) – AgentRun ID to update deploy to.
environment (Dict[str, str] | None) – Environment variables for the runtime.

Returns:

success: Whether deployment succeeded
agent_runtime_id: ID of the created runtime
agent_runtime_endpoint_id: ID of the created endpoint
agent_runtime_endpoint_name: Name of the endpoint
agent_runtime_public_endpoint_url: Public URL of the endpoint
status: Status of the deployment
request_id: Request ID for tracking

Return type:

Dictionary containing deployment results

async delete(agent_runtime_id)[source]

Delete an agent runtime on AgentRun.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime to delete.

Returns:

A dictionary containing the delete result with:

success (bool): Whether the operation was successful
message (str): Status message
agent_runtime_id (str): The ID of the deleted agent runtime
status (str): The status of the agent runtime
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async get_agent_runtime(agent_runtime_id, agent_runtime_version=None)[source]

Get agent runtime details.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
agent_runtime_version (str, optional) – The version of the agent runtime.

Returns:

A dictionary containing the agent runtime details with:

success (bool): Whether the operation was successful
data (dict): The agent runtime data
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async create_agent_runtime(agent_runtime_name, artifact_type, cpu, memory, port, code_configuration=None, description=None, environment_variables=None, execution_role_arn=None, log_configuration=None, network_configuration=None, session_concurrency_limit_per_instance=None, session_idle_timeout_seconds=None)[source]

Create an agent runtime on AgentRun.

Parameters:

agent_runtime_name (str) – The name of the agent runtime.
artifact_type (str) – The type of the artifact.
cpu (float) – The CPU allocated to the runtime.
memory (int) – The memory allocated to the runtime.
port (int) – The port on which the runtime will listen.
code_configuration (Optional[CodeConfig]) – Configuration for code-based runtimes.
description (Optional[str]) – Description of the agent runtime.
environment_variables (Optional[Dict[str, str]]) – Environment variables for the runtime.
execution_role_arn (Optional[str]) – The execution role ARN for accessing cloud services.
log_configuration (Optional[LogConfig]) – Configuration for logging.
network_configuration (Optional[NetworkConfig]) – Network configuration for the runtime, including: - network_mode: The network mode for the runtime - security_group_id: The security group ID for the runtime - vpc_id: The VPC ID for the runtime - vswitch_ids: List of vswitch IDs for the runtime
session_concurrency_limit_per_instance (Optional[int]) – Maximum concurrent sessions per instance.
session_idle_timeout_seconds (Optional[int]) – Maximum idle timeout for sessions.

Returns:

A dictionary containing the creation result with:

success (bool): Whether the operation was successful
agent_runtime_id (str): The ID of the created agent runtime
status (str): The status of the agent runtime
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async update_agent_runtime(agent_runtime_id, agent_runtime_name=None, artifact_type=None, cpu=None, memory=None, port=None, code_configuration=None, description=None, environment_variables=None, execution_role_arn=None, log_configuration=None, network_configuration=None, session_concurrency_limit_per_instance=None, session_idle_timeout_seconds=None)[source]

Update an agent runtime on AgentRun.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime to update.
agent_runtime_name (Optional[str]) – The name of the agent runtime.
artifact_type (Optional[str]) – The type of the artifact.
cpu (Optional[float]) – The CPU allocated to the runtime.
memory (Optional[int]) – The memory allocated to the runtime.
port (Optional[int]) – The port on which the runtime will listen.
code_configuration (Optional[CodeConfig]) – Configuration for code-based runtimes.
description (Optional[str]) – Description of the agent runtime.
environment_variables (Optional[Dict[str, str]]) – Environment variables for the runtime.
execution_role_arn (Optional[str]) – The execution role ARN for accessing cloud services.
log_configuration (Optional[LogConfig]) – Configuration for logging.
network_configuration (Optional[NetworkConfig]) – Network configuration for the runtime, including: - network_mode: The network mode for the runtime - security_group_id: The security group ID for the runtime - vpc_id: The VPC ID for the runtime - vswitch_ids: List of vswitch IDs for the runtime
session_concurrency_limit_per_instance (Optional[int]) – Maximum concurrent sessions per instance.
session_idle_timeout_seconds (Optional[int]) – Maximum idle timeout for sessions.

Returns:

A dictionary containing the update result with:

success (bool): Whether the operation was successful
agent_runtime_id (str): The ID of the updated agent runtime
status (str): The status of the agent runtime
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async create_agent_runtime_endpoint(agent_runtime_id, endpoint_config=None)[source]

Create an agent runtime endpoint.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
endpoint_config (Optional[EndpointConfig]) – Configuration for the endpoint, including: - agent_runtime_endpoint_name: The name of the endpoint - description: Description of the endpoint - target_version: Target version for the endpoint

Returns:

A dictionary containing the creation result with:

success (bool): Whether the operation was successful
agent_runtime_endpoint_id (str): The ID of the created endpoint
agent_runtime_endpoint_name (str): The name of the created endpoint
agent_runtime_public_endpoint_url (str): The public URL of the endpoint
status (str): The status of the endpoint
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async update_agent_runtime_endpoint(agent_runtime_id, agent_runtime_endpoint_id, endpoint_config=None)[source]

Update an agent runtime endpoint.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
agent_runtime_endpoint_id (str) – The ID of the agent runtime endpoint.
endpoint_config (Optional[EndpointConfig]) – Configuration for the endpoint, including: - agent_runtime_endpoint_name: The name of the endpoint - description: Description of the endpoint - target_version: Target version for the endpoint

Returns:

A dictionary containing the update result with:

success (bool): Whether the operation was successful
agent_runtime_endpoint_id (str): The ID of the updated endpoint
status (str): The status of the endpoint
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async get_agent_runtime_endpoint(agent_runtime_id, agent_runtime_endpoint_id)[source]

Get an agent runtime endpoint.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
agent_runtime_endpoint_id (str) – The ID of the agent runtime endpoint.

Returns:

A dictionary containing the endpoint details with:

success (bool): Whether the operation was successful
agent_runtime_endpoint_id (str): The ID of the endpoint
agent_runtime_endpoint_name (str): The name of the endpoint
agent_runtime_id (str): The ID of the agent runtime
agent_runtime_public_endpoint_url (str): The public URL of the endpoint
status (str): The status of the endpoint
status_reason (str): The reason for the status
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async delete_agent_runtime_endpoint(agent_runtime_id, agent_runtime_endpoint_id)[source]

Delete an agent runtime endpoint.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
agent_runtime_endpoint_id (str) – The ID of the agent runtime endpoint.

Returns:

A dictionary containing the delete result with:

success (bool): Whether the operation was successful
message (str): Status message
agent_runtime_endpoint_id (str): The ID of the deleted endpoint
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async publish_agent_runtime_version(agent_runtime_id, description=None)[source]

Publish an agent runtime version.

Parameters:

agent_runtime_id (str) – The ID of the agent runtime.
description (Optional[str]) – Description of the version.

Returns:

A dictionary containing the publish result with:

success (bool): Whether the operation was successful
agent_runtime_id (str): The ID of the agent runtime
agent_runtime_version (str): The published version
description (str): Description of the version
request_id (str): The request ID for tracking

Return type:

Dict[str, Any]

async stop(deploy_id, **kwargs)[source]

Stop AgentRun deployment by deleting it.

Parameters:

deploy_id (str) – AgentRun runtime ID (agent_runtime_id)
**kwargs – Additional parameters

Returns:

Dict with success status, message, and details

Return type:

agentscope_runtime.engine.deployers.cli_fc_deploy.main()[source]

Return type:: None

Deployers · Adapter¶

class agentscope_runtime.engine.deployers.adapter.protocol_adapter.ProtocolAdapter(**kwargs)[source]

Bases: ABC

__init__(**kwargs)[source]

abstract add_endpoint(app, func, **kwargs)[source]

Add an endpoint to the protocol adapter.

This method should be implemented by subclasses to provide protocol-specific endpoint addition functionality.

Parameters:

*args – Variable length argument list for endpoint configuration
**kwargs – Arbitrary keyword arguments for endpoint configuration
func (Callable)

Returns:

The result of adding the endpoint, implementation-dependent

Return type:

Any

class agentscope_runtime.engine.deployers.adapter.a2a.a2a_agent_adapter.A2AExecutor(func, **kwargs)[source]

Bases: AgentExecutor

Parameters:: func (Callable)

__init__(func, **kwargs)[source]

Parameters:: func (Callable)

async execute(context, event_queue)[source]

Execute the agent’s logic for a given request context.

The agent should read necessary information from the context and publish Task or Message events, or TaskStatusUpdateEvent / TaskArtifactUpdateEvent to the event_queue. This method should return once the agent’s execution for this request is complete or yields control (e.g., enters an input-required state).

Parameters:

context (RequestContext) – The request context containing the message, task ID, etc.
event_queue (EventQueue) – The queue to publish events to.

Return type:

None

async cancel(context, event_queue)[source]

Request the agent to cancel an ongoing task.

The agent should attempt to stop the task identified by the task_id in the context and publish a TaskStatusUpdateEvent with state TaskState.canceled to the event_queue.

Parameters:

context (RequestContext) – The request context containing the task ID to cancel.
event_queue (EventQueue) – The queue to publish the cancellation status update to.

Return type:

None

class agentscope_runtime.engine.deployers.adapter.a2a.a2a_protocol_adapter.A2AFastAPIDefaultAdapter(agent_name, agent_description, **kwargs)[source]

Bases: ProtocolAdapter

__init__(agent_name, agent_description, **kwargs)[source]

add_endpoint(app, func, **kwargs)[source]

Add an endpoint to the protocol adapter.

This method should be implemented by subclasses to provide protocol-specific endpoint addition functionality.

Parameters:

*args – Variable length argument list for endpoint configuration
**kwargs – Arbitrary keyword arguments for endpoint configuration
func (Callable)

Returns:

The result of adding the endpoint, implementation-dependent

Return type:

Any

get_agent_card(agent_name, agent_description)[source]

Parameters:

agent_name (str)
agent_description (str)

Return type:

AgentCard

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.a2a_message_to_agent_message(msg)[source]

Convert A2A Message object to AgentAPI Message object

Parameters:: msg (A2AMessage) – A2A protocol message object
Returns:: Converted internal Agent API message object
Return type:: AgentMessage

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.a2a_part_to_agent_content(part)[source]

Convert A2A protocol Part object to internal Content object

Parameters:: part (Part) – A2A protocol part object
Returns:: Converted internal content object
Return type:: Content
Raises:: ValueError – If the part type is unknown

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.a2a_sendparams_to_agent_request(params, stream, context_id=None)[source]

Convert a2a MessageSendParams to agent-api AgentRequest

Parameters:

params (dict) – MessageSendParams received from a2a protocol
stream (bool) – Whether this request is in stream mode (/message/send = False, /message/stream = True)
context_id (str, optional) – Context ID if message is appended to existing conversation

Returns:

Converted agent request object

Return type:

AgentRequest

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.a2a_taskqueryparams_to_agent_request(params, session_id=None)[source]

Convert TaskQueryParams to AgentRequest, only set session_id Other fields are controlled by AgentRequest default values

Parameters:

params (TaskQueryParams) – Task query parameters from a2a protocol
session_id (str, optional) – Session ID for the request

Returns:

Converted agent request object with only session_id set

Return type:

AgentRequest

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.agent_content_to_a2a_part(content)[source]

Convert internal Content object to A2A protocol Part object

Parameters:: content (Content) – Internal content object
Returns:: Converted A2A protocol part object
Return type:: Part
Raises:: ValueError – If the content type is unknown

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.agent_message_to_a2a_artifact(msg)[source]

Convert AgentAPI Message to a2a Artifact

Parameters:: msg (AgentMessage) – Agent API message object
Returns:: Converted A2A protocol artifact object
Return type:: Artifact

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.runstatus_to_a2a_taskstate(status)[source]

Map Internal RunStatus to a2a TaskState

Parameters:: status (str) – Internal run status string
Returns:: Mapped A2A task state
Return type:: TaskState

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.agent_response_to_a2a_task(resp)[source]

Convert AgentResponse object to a2a Task object.

Parameters:: resp (AgentResponse) – Internal agent response object
Returns:: Converted A2A protocol task object
Return type:: Task

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.response_to_task_status_update_event(response)[source]

Convert AgentResponse (internal response) to a2a TaskStatusUpdateEvent.

Parameters:: response (AgentResponse) – Internal agent response object
Returns:: Converted A2A protocol task status update event
Return type:: TaskStatusUpdateEvent

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.content_to_task_artifact_update_event(content, context_id='', task_id=None, append=False, last_chunk=False)[source]

Convert single Content (TextContent/ImageContent/DataContent) to TaskArtifactUpdateEvent. If delta=false, should not return task_artifact_update_event

Parameters:

content (Content) – SSE returned content, including delta and non-delta types
context_id (str) – Corresponds to agent api sessionId, needs external input
task_id (str | None) – Currently equivalent to msg_id, or not passed
append (bool) – Used to determine if current artifact is new or first
last_chunk (bool) – Used to determine if current content is the last one

Returns:

Converted A2A protocol task artifact update event

Return type:

TaskArtifactUpdateEvent

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.agent_role_to_a2a_role(role)[source]

Parameters:: role (str)

agentscope_runtime.engine.deployers.adapter.a2a.a2a_adapter_utils.agent_message_to_a2a_message(msg)[source]

Convert AgentAPI Message object to a2a protocol Message object

Parameters:: msg (AgentAPIMessage) – Agent API message object
Returns:: Converted A2A protocol message object
Return type:: A2AMessage

class agentscope_runtime.engine.deployers.adapter.responses.response_api_agent_adapter.ResponseAPIExecutor(func, **kwargs)[source]

Bases: object

Parameters:: func (Callable)

__init__(func, **kwargs)[source]

Parameters:: func (Callable)

async execute(request)[source]

Parameters:: request (Dict)

class agentscope_runtime.engine.deployers.adapter.responses.response_api_protocol_adapter.ResponseAPIDefaultAdapter(**kwargs)[source]

Bases: ProtocolAdapter

__init__(**kwargs)[source]

Return type:: None

add_endpoint(app, func, **kwargs)[source]

Add endpoint to FastAPI app.

Parameters:

app (FastAPI) – FastAPI application instance
func (Callable) – handler function
**kwargs – extra options

Return type:

None

Responses Adapter

Bidirectional protocol converter: Responses API ↔ Agent API

Conversion functions: 1. Responses API request → Agent API request 2. Agent API event → Responses API event 3. Support for streaming and non-streaming conversion

class agentscope_runtime.engine.deployers.adapter.responses.response_api_adapter_utils.ResponsesAdapter[source]

Bases: object

Bidirectional protocol converter: Responses API ↔ Agent API

Main functions: 1. Convert Responses API request → Agent API request 2. Convert Agent API event → Responses API event 3. Convert Responses API event stream → Agent API event stream 4. Handle various message types (text, tool calls, reasoning, etc.)

__init__()[source]

Return type:: None

convert_agent_response_to_responses(agent_response)[source]

Parameters:: agent_response (BaseResponse)

convert_status_to_responses(agent_status)[source]

Parameters:: agent_status (str)
Return type:: Literal[‘completed’, ‘failed’, ‘in_progress’, ‘cancelled’, ‘queued’, ‘incomplete’]

convert_responses_request_to_agent_request(responses_request)[source]

Convert Responses API request to Agent API request

Implement automatic assignment of fields with the same name, then explicitly handle different field names

Parameters:: responses_request (Dict[str, Any]) – OpenAI ResponseCreateParams
Returns:: Agent API request format
Return type:: AgentRequest

convert_agent_event_to_responses_event(agent_event)[source]

Convert Agent API event to Responses API stream event

Parameters:: agent_event (Event) – Agent API Event
Returns:: ResponseStreamEvent or None
Return type:: List[Annotated[ResponseAudioDeltaEvent | ResponseAudioDoneEvent | ResponseAudioTranscriptDeltaEvent | ResponseAudioTranscriptDoneEvent | ResponseCodeInterpreterCallCodeDeltaEvent | ResponseCodeInterpreterCallCodeDoneEvent | ResponseCodeInterpreterCallCompletedEvent | ResponseCodeInterpreterCallInProgressEvent | ResponseCodeInterpreterCallInterpretingEvent | ResponseCompletedEvent | ResponseContentPartAddedEvent | ResponseContentPartDoneEvent | ResponseCreatedEvent | ResponseErrorEvent | ResponseFileSearchCallCompletedEvent | ResponseFileSearchCallInProgressEvent | ResponseFileSearchCallSearchingEvent | ResponseFunctionCallArgumentsDeltaEvent | ResponseFunctionCallArgumentsDoneEvent | ResponseInProgressEvent | ResponseFailedEvent | ResponseIncompleteEvent | ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent | ResponseReasoningSummaryPartAddedEvent | ResponseReasoningSummaryPartDoneEvent | ResponseReasoningSummaryTextDeltaEvent | ResponseReasoningSummaryTextDoneEvent | ResponseReasoningTextDeltaEvent | ResponseReasoningTextDoneEvent | ResponseRefusalDeltaEvent | ResponseRefusalDoneEvent | ResponseTextDeltaEvent | ResponseTextDoneEvent | ResponseWebSearchCallCompletedEvent | ResponseWebSearchCallInProgressEvent | ResponseWebSearchCallSearchingEvent | ResponseImageGenCallCompletedEvent | ResponseImageGenCallGeneratingEvent | ResponseImageGenCallInProgressEvent | ResponseImageGenCallPartialImageEvent | ResponseMcpCallArgumentsDeltaEvent | ResponseMcpCallArgumentsDoneEvent | ResponseMcpCallCompletedEvent | ResponseMcpCallFailedEvent | ResponseMcpCallInProgressEvent | ResponseMcpListToolsCompletedEvent | ResponseMcpListToolsFailedEvent | ResponseMcpListToolsInProgressEvent | ResponseOutputTextAnnotationAddedEvent | ResponseQueuedEvent | ResponseCustomToolCallInputDeltaEvent | ResponseCustomToolCallInputDoneEvent, PropertyInfo(alias=’None’, format=None, format_template=’None’, discriminator=’type’)]] | None

convert_response_to_agent_events(response)[source]

Convert OpenAI Response object to Agent API Event list

Parameters:: response (Response) – OpenAI Response object
Returns:: Agent API Event list
Return type:: List[Event]

Deployers · Utils¶

Helpers for working with AgentApp objects inside deployers.

agentscope_runtime.engine.deployers.utils.app_runner_utils.ensure_runner_from_app(app)[source]

Return a Runner extracted from an AgentApp instance.

Builds the runner lazily if the app hasn’t initialized it yet.

Parameters:: app (Any)
Return type:: Optional[‘Runner’]

Deployment modes and configuration for unified FastAPI architecture.

class agentscope_runtime.engine.deployers.utils.deployment_modes.DeploymentMode(value)[source]

Bases: str, Enum

FastAPI application deployment modes.

DAEMON_THREAD = 'daemon_thread'

DETACHED_PROCESS = 'detached_process'

STANDALONE = 'standalone'

__format__(format_spec): Returns format using actual value type unless __str__ has been overridden.

Shared helpers for building detached deployment bundles.

agentscope_runtime.engine.deployers.utils.detached_app.build_detached_app(*, app=None, runner=None, entrypoint=None, requirements=None, extra_packages=None, output_dir=None, dockerfile_path=None, use_local_runtime=None, platform='unknown', **kwargs)[source]

Create a detached bundle directory ready for execution.

All temporary files are created in cwd/.agentscope_runtime/ by default.

Parameters:

app – AgentApp instance to deploy
runner – Runner instance to deploy
entrypoint (str | None) – Entrypoint specification (e.g., “app.py” or “app.py:handler”)
requirements (str | None | List[str]) – Additional pip requirements (string or list)
extra_packages (List[str] | None) – Additional Python packages to include
output_dir (str | None) – Output directory (creates temp dir if None)
dockerfile_path (str | None) – Optional custom Dockerfile path to include
use_local_runtime (bool | None) – If True, build and include local runtime wheel. If None (default), auto-detect based on version. Useful for development when runtime is not released.
platform (str)

Returns:

Tuple of (project_root_path, project_info)

Return type:

Tuple[str, ProjectInfo]

agentscope_runtime.engine.deployers.utils.detached_app.append_project_requirements(extraction_dir, additional_requirements, use_local_runtime=False)[source]

Append requirements to requirements.txt.

For dev versions or when use_local_runtime=True, builds a wheel from local source and places it in wheels/ subdirectory.

Parameters:

extraction_dir (Path) – Directory where requirements.txt will be written
additional_requirements (str | list | None) – Additional user requirements
use_local_runtime (bool | None) – If True, build and use local runtime wheel. Useful for development when runtime is not released.

Return type:

None

agentscope_runtime.engine.deployers.utils.detached_app.get_bundle_entry_script(bundle_dir)[source]

Parameters:: bundle_dir (str | Path | None)
Return type:: str

Project-based packaging utilities for AgentApp and Runner deployment.

This module provides packaging utilities that support: - Function-based AgentApp deployment with decorators - Runner-based deployment with entrypoint files - Entire project directory packaging - Smart dependency caching - CLI-style and object-style deployment patterns

agentscope_runtime.engine.deployers.utils.package.generate_build_directory(platform='unknown', workspace=None)[source]

Generate a platform-aware build directory with timestamp and random suffix.

Parameters:

platform (str) – Deployment platform (k8s, modelstudio, agentrun, local, etc.)
workspace (Path | None) – Custom workspace directory (defaults to DEFAULT_BUILD_WORKSPACE)

Returns:

Generated build directory path

Return type:

Path

Example

>>> build_dir = generate_build_directory("modelstudio")
>>> # Returns: .agentscope_runtime/builds/modelstudio_20251207_xxx

class agentscope_runtime.engine.deployers.utils.package.RuntimeParameter(*, name, type, default, help=None, cli_name=None)[source]

Bases: BaseModel

Configuration for a runtime parameter.

Parameters:

name (str)
type (str)
default (str | int | bool | float | None)
help (str | None)
cli_name (str | None)

name: str

type: str

default: str | int | bool | float | None

help: str | None

cli_name: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.package.EntrypointInfo(*, module_name, object_name, object_type, host='0.0.0.0', port=8090, extra_parameters=[])[source]

Bases: BaseModel

Information about the generated entrypoint.

Parameters:

module_name (str)
object_name (str)
object_type (str)
host (str)
port (int)
extra_parameters (List[RuntimeParameter])

module_name: str

object_name: str

object_type: str

host: str

port: int

extra_parameters: List[RuntimeParameter]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.package.ProjectInfo(*, project_dir, entrypoint_file, entrypoint_handler, handler_type=None, is_directory_entrypoint=False)[source]

Bases: BaseModel

Information about a project to be packaged.

Parameters:

project_dir (str)
entrypoint_file (str)
entrypoint_handler (str)
handler_type (str | None)
is_directory_entrypoint (bool)

project_dir: str

entrypoint_file: str

entrypoint_handler: str

handler_type: str | None

is_directory_entrypoint: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

agentscope_runtime.engine.deployers.utils.package.project_dir_extractor(app=None, runner=None)[source]

Extract project directory information from app or runner object.

This function inspects the call stack to find where the app or runner was defined and extracts the project root directory and object name.

Parameters:

app – AgentApp instance (optional)
runner – Runner instance (optional)

Returns:

ProjectInfo with project directory, entrypoint file, and handler name

Raises:

ValueError – If neither app nor runner is provided or project dir
cannot be determined –

Return type:

ProjectInfo

agentscope_runtime.engine.deployers.utils.package.parse_entrypoint(spec)[source]

Parse entrypoint specification into ProjectInfo.

Supported formats: - “app.py” - File with default handler name “app” - “app.py:my_handler” - File with specific handler name - “project_dir/” - Directory (will auto-detect entrypoint)

Parameters:

spec (str) – Entrypoint specification string

Returns:

ProjectInfo with parsed information

Raises:

ValueError – If specification format is invalid or file/dir doesn’t
exist –

Return type:

ProjectInfo

agentscope_runtime.engine.deployers.utils.package.generate_main_template(entrypoint_info)[source]

Generate main.py template based on object type using Jinja2 templates.

Parameters:

entrypoint_info (EntrypointInfo) – Information about the entrypoint

Returns:

String content for main.py

Raises:

ValueError – If object_type is not supported
RuntimeError – If template rendering fails

Return type:

agentscope_runtime.engine.deployers.utils.package.package_code(source_dir, output_zip, ignore_patterns=None)[source]

Package project source code into a zip file.

Parameters:

source_dir (Path) – Source directory to package
output_zip (Path) – Output zip file path
ignore_patterns (List[str] | None) – Optional ignore patterns (uses defaults if None)

Return type:

None

agentscope_runtime.engine.deployers.utils.package.package(app=None, runner=None, entrypoint=None, output_dir=None, host='0.0.0.0', port=8090, extra_parameters=None, requirements=None, platform='unknown', **kwargs)[source]

Package an AgentApp or Runner for deployment.

This function supports two deployment patterns: 1. Object-style: package(app=my_app) or package(runner=my_runner) 2. Entrypoint-style: package(entrypoint=”app.py”) or package( entrypoint=”project_dir/”)

For object-style deployment, this function will: 1. Extract the project directory containing the app/runner 2. Generate a new main.py that imports and runs the app/runner 3. Package the project with the generated main.py as entrypoint

Build directory naming: - When output_dir=None (default), creates workspace directory with

platform-aware naming

Directory format: cwd/.agentscope_runtime/builds/<platform>_
<timestamp>_<code>/
Explicit output_dir uses the provided path

Parameters:

app – AgentApp instance (for object-style deployment)
runner – Runner instance (for object-style deployment)
entrypoint (str | None) – Entrypoint specification (for CLI-style deployment)
output_dir (str | None) – Output directory (creates temp dir if None)
host (str) – Default host for the service (default: “0.0.0.0”)
port (int) – Default port for the service (default: 8090)
extra_parameters (List[RuntimeParameter] | None) – Additional runtime parameters to expose via CLI
requirements (List[str] | None) – Additional pip requirements
platform (str) – Deployment platform (k8s, modelstudio, agentrun, local)
**kwargs – Additional keyword arguments (ignored)

Returns:

Tuple of (package_path, project_info) - package_path: Path to the deployment package directory - project_info: ProjectInfo with project metadata

Raises:

ValueError – If neither app/runner nor entrypoint is provided
RuntimeError – If packaging fails

Return type:

Tuple[str, ProjectInfo]

Example

>>> # Package with extra parameters
>>> extra_params = [
...     RuntimeParameter(
...         name="log_level",
...         type="str",
...         default="info",
...         help="Logging level"
...     ),
...     RuntimeParameter(
...         name="workers",
...         type="int",
...         default=4,
...         help="Number of worker threads"
...     ),
... ]
>>> package(app=my_app, extra_parameters=extra_params, platform="k8s")

Utilities for packaging a local Python project into a distributable wheel that can be uploaded and deployed by various deployers.

This module extracts and generalizes logic from the legacy test script tests/integrated/test_bailian_fc_deploy/deploy_builder.py so that production deployers can reuse the behaviour in a structured way.

agentscope_runtime.engine.deployers.utils.wheel_packager.get_user_bundle_appdir(build_root, user_project_dir)[source]

Parameters:

build_root (Path)
user_project_dir (Path)

Return type:

Path

agentscope_runtime.engine.deployers.utils.wheel_packager.generate_wrapper_project(build_root, user_project_dir, start_cmd, deploy_name, telemetry_enabled=True, requirements=None)[source]

Create a wrapper project under build_root, embedding user project under user_bundle/<project_basename>. Returns: (wrapper_project_dir, dist_dir)

Parameters:

build_root (Path)
user_project_dir (Path)
start_cmd (str)
deploy_name (str)
telemetry_enabled (bool)
requirements (str | None | List[str])

Return type:

Tuple[Path, Path]

agentscope_runtime.engine.deployers.utils.wheel_packager.build_wheel(project_dir)[source]

Build a wheel inside an isolated virtual environment to avoid PEP 668 issues. Returns the path to the built wheel.

Parameters:: project_dir (Path)
Return type:: Path

agentscope_runtime.engine.deployers.utils.wheel_packager.default_deploy_name()[source]

Return type:: str

class agentscope_runtime.engine.deployers.utils.docker_image_utils.docker_image_builder.RegistryConfig(*, registry_url='', username=None, password=None, namespace='agentscope-runtime', image_pull_secret=None)[source]

Bases: BaseModel

Container registry configuration

Parameters:

registry_url (str)
username (str)
password (str)
namespace (str | None)
image_pull_secret (str)

registry_url: str

username: str

password: str

namespace: str | None

image_pull_secret: str

get_full_url()[source]

Return type:: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.docker_image_utils.docker_image_builder.BuildConfig(*, no_cache=False, quiet=False, build_args={}, platform=None, target=None, source_updated=False)[source]

Bases: BaseModel

Configuration for Docker image building

Parameters:

no_cache (bool)
quiet (bool)
build_args (Dict[str, str])
platform (str | None)
target (str | None)
source_updated (bool)

no_cache: bool

quiet: bool

build_args: Dict[str, str]

platform: str | None

target: str | None

source_updated: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.docker_image_utils.docker_image_builder.DockerImageBuilder[source]

Bases: object

Responsible solely for building and managing Docker images. Separated from project packaging for better separation of concerns.

__init__()[source]: Initialize Docker image builder.

static get_full_name(image_name, image_tag='latest')[source]

Parameters:

image_name (str)
image_tag (str)

build_image(build_context, image_name, image_tag='latest', dockerfile_path=None, config=None, source_updated=False)[source]

Build Docker image from build context.

Parameters:

build_context (str) – Path to build context directory
image_name (str) – Name for the Docker image
image_tag (str) – Tag for the Docker image
dockerfile_path (str | None) – Optional path to Dockerfile (defaults to Dockerfile in context)
config (BuildConfig | None) – Build configuration
source_updated (bool) – Optional flag to determine if source image should be updated.

Returns:

Full image name with tag

Return type:

Raises:

subprocess.CalledProcessError – If docker build fails
ValueError – If build context doesn’t exist

tag_image(image_name, registry_config=None)[source]

Tag image with registry info.

Parameters:

image_name (str) – Full image name to push
registry_config (RegistryConfig | None) – Optional registry config (uses instance config if None)

Returns:

Full image name with url

Return type:

registry_image_name

push_image(image_name, registry_config=None, quiet=False)[source]

Push image to registry.

Parameters:

image_name (str) – Full image name to push
registry_config (RegistryConfig | None) – Optional registry config (uses instance config if None)
quiet (bool) – Whether to suppress output

Returns:

Full image name that was pushed

Return type:

Raises:

subprocess.CalledProcessError – If push fails
ValueError – If no registry configuration is available

build_and_push(build_context, image_name, image_tag='latest', dockerfile_path=None, build_config=None, registry_config=None, source_updated=False)[source]

Build and push image in one operation.

Parameters:

build_context (str) – Path to build context directory
image_name (str) – Name for the Docker image
image_tag (str) – Tag for the Docker image
dockerfile_path (str | None) – Optional path to Dockerfile
build_config (BuildConfig | None) – Build configuration
registry_config (RegistryConfig | None) – Registry configuration
source_updated (bool) – Whether the source image was updated or not

Returns:

Full registry image name

Return type:

remove_image(image_name, force=False, quiet=True)[source]

Remove Docker image.

Parameters:

image_name (str) – Name of image to remove
force (bool) – Force removal
quiet (bool) – Suppress output

Returns:

True if successful

Return type:

bool

get_image_info(image_name)[source]

Get information about a Docker image.

Parameters:: image_name (str) – Name of the Docker image
Returns:: Image information from docker inspect
Return type:: Dict
Raises:: ValueError – If image not found or info invalid

image_exists(image_name)[source]

Check if Docker image exists locally.

Parameters:: image_name (str) – Name of image to check
Returns:: True if image exists
Return type:: bool

class agentscope_runtime.engine.deployers.utils.docker_image_utils.dockerfile_generator.DockerfileConfig(*, base_image='python:3.10-slim-bookworm', port=8000, working_dir='/app', user='appuser', additional_packages=[], env_vars={}, startup_command=None, health_check_endpoint='/health', custom_template=None, platform=None)[source]

Bases: BaseModel

Configuration for Dockerfile generation

Parameters:

base_image (str)
port (int)
working_dir (str)
user (str)
additional_packages (List[str])
env_vars (Dict[str, str])
startup_command (str | None)
health_check_endpoint (str)
custom_template (str | None)
platform (str | None)

base_image: str

port: int

working_dir: str

user: str

additional_packages: List[str]

env_vars: Dict[str, str]

startup_command: str | None

health_check_endpoint: str

custom_template: str | None

platform: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.docker_image_utils.dockerfile_generator.DockerfileGenerator[source]

Bases: object

Responsible for generating Dockerfiles from templates. Separated from image building for better modularity.

DEFAULT_TEMPLATE = '# Use official Python runtime as base image\nFROM --platform={platform} {base_image}\n\n# Set working directory in container\nWORKDIR /app\n\n# Set environment variables\nENV PYTHONDONTWRITEBYTECODE=1\nENV PYTHONUNBUFFERED=1\n\n# Configure package sources for better performance\nRUN rm -f /etc/apt/sources.list.d/*.list\n\n# add aliyun mirrors\nRUN echo "deb https://mirrors.aliyun.com/debian/ bookworm main contrib " "non-free non-free-firmware" > /etc/apt/sources.list && echo "deb https://mirrors.aliyun.com/debian/ bookworm-updates main " "contrib non-free non-free-firmware" >> /etc/apt/sources.list && echo "deb https://mirrors.aliyun.com/debian-security/ " "bookworm-security main contrib non-free " "non-free-firmware" >> /etc/apt/sources.list\n\n# replace debian to aliyun\nRUN mkdir -p /etc/apt/sources.list.d && cat > /etc/apt/sources.list.d/debian.sources <<\'EOF\'\nEOF\n\n# Clean up package lists\nRUN rm -rf /var/lib/apt/lists/*\n\n# Install system dependencies\nRUN apt-get update && apt-get install -y \\\n gcc \\\n curl \\\n{additional_packages_section} && rm -rf /var/lib/apt/lists/*\n\n# Copy project files\nCOPY . {working_dir}/\n\n# Install Python dependencies\nRUN pip install --no-cache-dir --upgrade pip\nRUN if [ -f requirements.txt ]; then \\\n pip install --no-cache-dir -r requirements.txt \\\n -i https://pypi.tuna.tsinghua.edu.cn/simple; fi\n\n# Create non-root user for security\nRUN adduser --disabled-password --gecos \'\' {user} && \\\n chown -R {user} {working_dir}\nUSER {user}\n\n{env_vars_section}\n# Expose port\nEXPOSE {port}\n\n# Health check\nHEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \\\n CMD curl -f http://localhost:{port}{health_check_endpoint} || exit 1\n\n# Command to run the application\n{startup_command_section}'

__init__()[source]

Return type:: None

temp_files: List[str]

generate_dockerfile_content(config)[source]

Generate Dockerfile content from configuration.

Parameters:: config (DockerfileConfig) – Dockerfile configuration
Returns:: Generated Dockerfile content
Return type:: str

create_dockerfile(config, output_dir=None)[source]

Create Dockerfile in specified directory.

Parameters:

config (DockerfileConfig) – Dockerfile configuration
output_dir (str | None) – Directory to create Dockerfile (temp dir if None)

Returns:

Path to created Dockerfile

Return type:

validate_config(config)[source]

Validate Dockerfile configuration.

Parameters:: config (DockerfileConfig) – Configuration to validate
Returns:: True if valid
Return type:: bool
Raises:: ValueError – If configuration is invalid

cleanup()[source]: Clean up temporary files

__enter__()[source]: Context manager entry

__exit__(exc_type, exc_val, exc_tb)[source]: Context manager exit with cleanup

class agentscope_runtime.engine.deployers.utils.docker_image_utils.image_factory.ImageConfig(*, requirements=None, extra_packages=None, build_context_dir=None, endpoint_path='/process', protocol_adapters=None, custom_endpoints=None, base_image='python:3.10-slim-bookworm', port=8000, env_vars=<factory>, startup_command=None, host='0.0.0.0', embed_task_processor=False, extra_startup_args=<factory>, no_cache=False, quiet=False, build_args={}, platform=None, image_name=('agent', ), image_tag=None, registry_config=None, push_to_registry=False)[source]

Bases: BaseModel

Complete configuration for building a Runner image

Parameters:

requirements (List[str] | None)
extra_packages (List[str] | None)
build_context_dir (str | None)
endpoint_path (str)
protocol_adapters (List | None)
custom_endpoints (List[Dict] | None)
base_image (str)
port (int)
env_vars (Dict[str, str])
startup_command (str | None)
host (str)
embed_task_processor (bool)
extra_startup_args (Dict[str, str | int | bool])
no_cache (bool)
quiet (bool)
build_args (Dict[str, str])
platform (str | None)
image_name (str | None)
image_tag (str | None)
registry_config (RegistryConfig | None)
push_to_registry (bool)

requirements: List[str] | None

extra_packages: List[str] | None

build_context_dir: str | None

endpoint_path: str

protocol_adapters: List | None

custom_endpoints: List[Dict] | None

base_image: str

port: int

env_vars: Dict[str, str]

startup_command: str | None

host: str

embed_task_processor: bool

extra_startup_args: Dict[str, str | int | bool]

no_cache: bool

quiet: bool

build_args: Dict[str, str]

platform: str | None

image_name: str | None

image_tag: str | None

registry_config: RegistryConfig | None

push_to_registry: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agentscope_runtime.engine.deployers.utils.docker_image_utils.image_factory.ImageFactory[source]

Bases: object

Factory class for building Runner Docker images. Coordinates ProjectPackager, DockerfileGenerator, and DockerImageBuilder.

__init__()[source]: Initialize the Runner image factory.

build_image(app=None, runner=None, entrypoint=None, requirements=None, extra_packages=None, base_image='python:3.10-slim-bookworm', image_name='agent', image_tag=None, registry_config=None, push_to_registry=False, protocol_adapters=None, custom_endpoints=None, host='0.0.0.0', embed_task_processor=True, extra_startup_args=None, use_cache=True, **kwargs)[source]

Simplified interface for building Runner images.

All temporary files are created in cwd/.agentscope_runtime/ by default.

Parameters:

app – agent app object
runner (Runner | None) – Runner object
entrypoint (str | None) – Entrypoint specification (e.g., “app.py” or “app.py:handler”)
requirements (str | None | List[str]) – Python requirements
extra_packages (List[str] | None) – Additional files to include
base_image (str) – Docker base image
image_name (str) – Docker image name
image_tag (str | None) – Optional image tag
registry_config (RegistryConfig | None) – Optional registry config
push_to_registry (bool) – Whether to push to registry
protocol_adapters (List | None) – Protocol adapters
custom_endpoints (List[Dict] | None) – Custom endpoints from agent app
host (str) – Host to bind to (default: 0.0.0.0 for containers)
embed_task_processor (bool) – Whether to embed task processor
extra_startup_args (Dict[str, str | int | bool] | None) – Additional startup arguments
use_cache (bool) – Enable build cache (default: True)
**kwargs – Additional configuration options

Returns:

Built image name

Return type:

cleanup()[source]: Clean up all temporary resources

__enter__()[source]: Context manager entry

__exit__(exc_type, exc_val, exc_tb)[source]: Context manager exit with cleanup

async agentscope_runtime.engine.deployers.utils.service_utils.fastapi_factory.error_stream(e)[source]

class agentscope_runtime.engine.deployers.utils.service_utils.fastapi_factory.FastAPIAppFactory[source]

Bases: object

Factory for creating FastAPI applications with unified architecture.

static create_app(func=None, runner=None, endpoint_path='/process', request_model=None, response_type='sse', stream=True, before_start=None, after_finish=None, mode=DeploymentMode.DAEMON_THREAD, protocol_adapters=None, custom_endpoints=None, broker_url=None, backend_url=None, enable_embedded_worker=False, app_kwargs=None, **kwargs)[source]

Create a FastAPI application with unified architecture.

Parameters:

func (Callable | None) – Custom processing function
runner (Any | None) – Runner instance (for DAEMON_THREAD mode)
endpoint_path (str) – API endpoint path for the processing function
request_model (Type | None) – Pydantic model for request validation
response_type (str) – Response type - “json”, “sse”, or “text”
stream (bool) – Enable streaming responses
before_start (Callable | None) – Callback function called before server starts
after_finish (Callable | None) – Callback function called after server finishes
mode (DeploymentMode) – Deployment mode
protocol_adapters (list[ProtocolAdapter] | None) – Protocol adapters
custom_endpoints (List[Dict] | None) – List of custom endpoint configurations
broker_url (str | None) – Celery broker URL
backend_url (str | None) – Celery backend URL
enable_embedded_worker (bool) – Whether to run embedded Celery worker
app_kwargs (Dict | None) – Additional keyword arguments for the FastAPI app
**kwargs (Any) – Additional keyword arguments

Returns:

FastAPI application instance

Return type:

FastAPI

FastAPI templates management and rendering.

class agentscope_runtime.engine.deployers.utils.service_utils.fastapi_templates.FastAPITemplateManager[source]

Bases: object

Manager for FastAPI deployment templates.

__init__()[source]: Initialize template manager.

render_standalone_template(agent_name, endpoint_path='/process', deployment_mode=DeploymentMode.STANDALONE, protocol_adapters=None, **kwargs)[source]

Render the standalone deployment template.

Parameters:

agent_name (str) – Name of the agent variable
endpoint_path (str) – API endpoint path
deployment_mode (str) – Deployment mode (standalone or detached_process)
protocol_adapters (str | None) – Protocol adapters code string
**kwargs – Additional template variables

Returns:

Rendered template content

Return type:

render_detached_script_template(endpoint_path='/process', host='127.0.0.1', port=8000, stream_enabled=True, response_type='sse', runner_code='', func_code='', protocol_adapters=None, **kwargs)[source]

Render the detached process script template.

Parameters:

endpoint_path (str) – API endpoint path
host (str) – Host to bind to
port (int) – Port to bind to
stream_enabled (bool) – Enable streaming responses
response_type (str) – Response type
runner_code (str) – Code to setup runner
func_code (str) – Code to setup custom function
protocol_adapters (str | None) – Protocol adapters code string
**kwargs – Additional template variables

Returns:

Rendered template content

Return type:

render_template_from_string(template_string, **variables)[source]

Render a template from string.

Parameters:

template_string (str) – Template content as string
**variables – Template variables

Returns:

Rendered template content

Return type:

get_template_list()[source]

Get list of available templates.

Returns:: List of template filenames
Return type:: list

validate_template_variables(template_name, variables)[source]

Validate template variables.

Parameters:

template_name (str) – Name of the template
variables (Dict[str, Any]) – Variables to validate

Returns:

Dictionary with ‘missing’ and ‘extra’ keys containing lists

Return type:

Dict[str, list]

class agentscope_runtime.engine.deployers.utils.service_utils.process_manager.ProcessManager(shutdown_timeout=30)[source]

Bases: object

Manager for detached process lifecycle.

Parameters:: shutdown_timeout (int)

__init__(shutdown_timeout=30)[source]

Initialize process manager.

Parameters:: shutdown_timeout (int) – Timeout in seconds for graceful shutdown

async start_detached_process(script_path, host='127.0.0.1', port=8000, env=None)[source]

Start a detached process running the given script.

Parameters:

script_path (str) – Path to the Python script to run
host (str) – Host to bind to
port (int) – Port to bind to
env (dict | None) – Additional environment variables

Returns:

Process PID

Raises:

RuntimeError – If process creation fails

Return type:

async stop_process_gracefully(pid, timeout=None)[source]

Stop a process gracefully.

Parameters:

pid (int) – Process ID to stop
timeout (int | None) – Timeout for graceful shutdown (uses default if None)

Returns:

True if process was stopped successfully

Raises:

RuntimeError – If process termination fails

Return type:

bool

is_process_running(pid)[source]

Check if a process is running.

Parameters:: pid (int) – Process ID to check
Returns:: True if process is running
Return type:: bool

create_pid_file(pid, file_path)[source]

Create a PID file.

Parameters:

pid (int) – Process ID
file_path (str) – Path to PID file

Raises:

OSError – If file creation fails

read_pid_file(file_path)[source]

Read PID from file.

Parameters:: file_path (str) – Path to PID file
Returns:: Process ID or None if file doesn’t exist or is invalid
Return type:: int | None

cleanup_pid_file(file_path)[source]

Remove PID file.

Parameters:: file_path (str) – Path to PID file

async find_process_by_port(port)[source]

Find process listening on a specific port.

Parameters:: port (int) – Port number
Returns:: Process ID or None if not found
Return type:: int | None

get_process_info(pid)[source]

Get information about a process.

Parameters:: pid (int) – Process ID
Returns:: Dictionary with process information or None if process not found
Return type:: dict | None

async wait_for_port(host, port, timeout=30)[source]

Wait for a service to become available on a port.

Parameters:

host (str) – Host to check
port (int) – Port to check
timeout (int) – Maximum time to wait

Returns:

True if service becomes available, False if timeout

Return type:

bool

get_process_logs(max_lines=50)[source]

Get the last N lines of process logs.

Parameters:: max_lines (int) – Maximum number of lines to return
Returns:: Log content as string
Return type:: str

cleanup_log_file(keep_file=False)[source]

Clean up log file.

Parameters:: keep_file (bool) – If True, keep the log file on disk but close the handle. If False, delete the log file.

static cleanup_old_logs(max_age_hours=24)[source]

Clean up old log files.

Parameters:: max_age_hours (int) – Remove log files older than this many hours

Services · 核心¶

class agentscope_runtime.engine.services.base.Service[source]

Bases: ABC

Abstract base class for services.

This class defines the interface that all services must implement.

abstract async start()[source]

Starts the service, initializing any necessary resources or connections.

Return type:: None

abstract async stop()[source]

Stops the service, releasing any acquired resources.

Return type:: None

abstract async health()[source]

Checks the health of the service.

Returns:: True if the service is healthy, False otherwise.
Return type:: bool

class agentscope_runtime.engine.services.base.ServiceLifecycleManagerMixin[source]

Bases: object

Mixin class that provides async lifecycle manager functionality for services.

This mixin can be used with any class that implements the Service interface.

async __aenter__()[source]: Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)[source]: Async context manager exit.

class agentscope_runtime.engine.services.base.ServiceWithLifecycleManager[source]

Bases: Service, ServiceLifecycleManagerMixin

Base class for services that want async lifecycle manager functionality.

This class combines the Service interface with the context manager mixin, providing a convenient base class for most service implementations.

Note: This is an abstract base class. Subclasses must implement the abstract methods from the Service class.

abstract async start()[source]

Starts the service, initializing any necessary resources or connections.

Return type:: None

abstract async stop()[source]

Stops the service, releasing any acquired resources.

Return type:: None

abstract async health()[source]

Checks the health of the service.

Returns:: True if the service is healthy, False otherwise.
Return type:: bool

Services · Agent State¶

class agentscope_runtime.engine.services.agent_state.state_service.StateService[source]

Bases: ServiceWithLifecycleManager

Abstract base class for agent state management services.

Stores and manages agent states organized by user_id, session_id, and round_id. Supports saving, retrieving, listing, and deleting states.

async start()[source]

Starts the service, initializing any necessary resources or connections.

Return type:: None

async stop()[source]

Stops the service, releasing any acquired resources.

Return type:: None

abstract async save_state(user_id, state, session_id=None, round_id=None)[source]

Save serialized state data for a specific user/session.

If round_id is provided, store the state in that round. If round_id is None, append as a new round with automatically assigned round_id.

Parameters:

user_id (str) – The unique ID of the user.
state (Dict[str, Any]) – A dictionary representing serialized agent state.
session_id (str | None) – Optional session/conversation ID. Defaults to “default”.
round_id (int | None) – Optional conversation round number.

Returns:

The round_id in which the state was saved.

Return type:

abstract async export_state(user_id, session_id=None, round_id=None)[source]

Retrieve serialized state data for a user/session.

If round_id is provided, return that round’s state. If round_id is None, return the latest round’s state.

Parameters:

user_id (str) – The unique ID of the user.
session_id (str | None) – Optional session/conversation ID.
round_id (int | None) – Optional round number.

Returns:

A dictionary representing the agent state, or None if not found.

Return type:

Dict[str, Any] | None

class agentscope_runtime.engine.services.agent_state.state_service.InMemoryStateService[source]

Bases: StateService

In-memory implementation of StateService using dictionaries for sparse round storage.

Multiple users, sessions, and non-contiguous round IDs are supported.
If round_id is None when saving, a new round is appended automatically.
If round_id is None when exporting, the latest round is returned.

__init__()[source]

Return type:: None

async start()[source]

Initialize the in-memory store.

Return type:: None

async stop()[source]

Clear all in-memory state data.

Return type:: None

async health()[source]

Service health check.

Return type:: bool

async save_state(user_id, state, session_id=None, round_id=None)[source]

Save serialized state in sparse dict storage.

If round_id is None, a new round_id will be assigned as (max existing round_id + 1) or 1 if none exist. Otherwise, the given round_id will be overwritten.

Returns:

The round_id where the state was saved.

Parameters:

user_id (str)
state (Dict[str, Any])
session_id (str | None)
round_id (int | None)

Return type:

async export_state(user_id, session_id=None, round_id=None)[source]

Retrieve state data for given user/session/round.

If round_id is None: return the latest round. If round_id is provided: return that round’s state.

Returns:

Dictionary representing the agent state, or None if not found.

Parameters:

user_id (str)
session_id (str | None)
round_id (int | None)

Return type:

Dict[str, Any] | None

class agentscope_runtime.engine.services.agent_state.redis_state_service.RedisStateService(redis_url='redis://localhost:6379/0', redis_client=None)[source]

Bases: StateService

Redis-based implementation of StateService.

Stores agent states in Redis using a hash per (user_id, session_id), with round_id as the hash field and serialized state as the value.

Parameters:

redis_url (str)
redis_client (Redis | None)

__init__(redis_url='redis://localhost:6379/0', redis_client=None)[source]

Parameters:

redis_url (str)
redis_client (Redis | None)

async start()[source]

Initialize the Redis connection.

Return type:: None

async stop()[source]

Close the Redis connection.

Return type:: None

async health()[source]

Service health check.

Return type:: bool

async save_state(user_id, state, session_id=None, round_id=None)[source]

Save serialized state data for a specific user/session.

If round_id is provided, store the state in that round. If round_id is None, append as a new round with automatically assigned round_id.

Parameters:

user_id (str) – The unique ID of the user.
state (Dict[str, Any]) – A dictionary representing serialized agent state.
session_id (str | None) – Optional session/conversation ID. Defaults to “default”.
round_id (int | None) – Optional conversation round number.

Returns:

The round_id in which the state was saved.

Return type:

async export_state(user_id, session_id=None, round_id=None)[source]

Retrieve serialized state data for a user/session.

If round_id is provided, return that round’s state. If round_id is None, return the latest round’s state.

Parameters:

user_id (str) – The unique ID of the user.
session_id (str | None) – Optional session/conversation ID.
round_id (int | None) – Optional round number.

Returns:

A dictionary representing the agent state, or None if not found.

Return type:

Dict[str, Any] | None

Services · Memory¶

class agentscope_runtime.engine.services.memory.memory_service.MemoryService[source]

Bases: ServiceWithLifecycleManager

Used to store and retrieve long memory from the database or in-memory. The memory is organized by the user id at top level, under which there are two different memory manage strategies, - one is the message grouped by the session id, the session id is under the user id, - the other is the message grouped by the user id only

abstract async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The messages to add.
session_id (str | None) – The session id, which is optional.

Return type:

None

async stop()[source]: Stops the service, releasing any acquired resources.

async start()[source]: Starts the service, initializing any necessary resources or connections.

abstract async search_memory(user_id, messages, filters=FieldInfo(annotation=NoneType, required=False, default=None, description='Associated filters for the messages, such as top_k, score etc.'))[source]

Searches messages from the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The user query or the query with history messages, both in the format of list of messages. If messages is a list, the search will be based on the content of the last message.
filters (Dict[str, Any] | None) – The filters used to search memory

Return type:

abstract async list_memory(user_id, filters=FieldInfo(annotation=NoneType, required=False, default=None, description='Associated filters for the messages, such as top_k, score etc.'))[source]

Lists the memory items for a given user with filters, such as page_num, page_size, etc.

Parameters:

user_id (str) – The user id.
filters (Dict[str, Any] | None) – The filters for the memory items.

Return type:

abstract async delete_memory(user_id, session_id=None)[source]

Deletes the memory items for a given user with certain session id, or all the memory items for a given user.

Parameters:

user_id (str)
session_id (str | None)

Return type:

None

class agentscope_runtime.engine.services.memory.memory_service.InMemoryMemoryService[source]

Bases: MemoryService

An in-memory implementation of the memory service.

__init__()[source]

Initializes the InMemorySessionHistoryService.

Return type:: None

async start()[source]

Initialize the in-memory store.

Return type:: None

async stop()[source]

Stops the service.

Return type:: None

async health()[source]

Checks the health of the service.

Return type:: bool

async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the in-memory store.

Parameters:

user_id (str) – The user’s unique identifier.
messages (list) – A list of messages to be added.
session_id (str | None) – An optional session identifier. If not provided,
used. (a default session is)

Return type:

None

async search_memory(user_id, messages, filters=None)[source]

Searches messages from the in-memory store for a specific user: based on keywords.

Parameters:

user_id (str) – The user’s unique identifier.
messages (list) – A list of messages, where the last message’s content is used as the search query.
filters (Dict[str, Any] | None) – Optional filters to apply, such as ‘top_k’ to limit the number of returned messages.

Returns:

A list of matching messages from the store.

Return type:

async get_query_text(message)[source]

Gets the query text from the messages.

Parameters:: message (Message) – A list of messages.
Returns:: The query text.
Return type:: str

async list_memory(user_id, filters=None)[source]

Lists messages from the in-memory store with pagination support.

Parameters:

user_id (str) – The user’s unique identifier.
filters (Dict[str, Any] | None) – Optional filters for pagination, including ‘page_num’ and ‘page_size’.

Returns:

A paginated list of messages.

Return type:

async delete_memory(user_id, session_id=None)[source]

Deletes messages from the in-memory store.

Parameters:

user_id (str) – The user’s unique identifier.
session_id (str | None) – If provided, only deletes the messages for that session. Otherwise, deletes all messages for the user.

Return type:

None

class agentscope_runtime.engine.services.memory.mem0_memory_service.Mem0MemoryService(**kwargs)[source]

Bases: MemoryService

Memory service that uses mem0 to store and retrieve memories. To get the api key, please refer to the following link: https://docs.mem0.ai/platform/quickstart

__init__(**kwargs)[source]

async static get_query_text(message)[source]

Get the query text from the message.

Parameters:: message (Message)
Return type:: str

static transform_message(message)[source]

Parameters:: message (Message)
Return type:: dict

async transform_messages(messages)[source]

Parameters:: messages (List[Message])
Return type:: List[dict]

async start()[source]: Starts the service, initializing any necessary resources or connections.

async stop()[source]: Stops the service, releasing any acquired resources.

async health()[source]

Checks the health of the service.

Returns:: True if the service is healthy, False otherwise.

async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The messages to add.
session_id (str | None) – The session id, which is optional.

async search_memory(user_id, messages, filters=None)[source]

Searches messages from the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The user query or the query with history messages, both in the format of list of messages. If messages is a list, the search will be based on the content of the last message.
filters (Dict[str, Any] | None) – The filters used to search memory

Return type:

async list_memory(user_id, filters=None)[source]

Lists the memory items for a given user with filters, such as page_num, page_size, etc.

Parameters:

user_id (str) – The user id.
filters (Dict[str, Any] | None) – The filters for the memory items.

Return type:

async delete_memory(user_id, session_id=None)[source]

Deletes the memory items for a given user with certain session id, or all the memory items for a given user.

Parameters:

user_id (str)
session_id (str | None)

Return type:

None

class agentscope_runtime.engine.services.memory.redis_memory_service.RedisMemoryService(redis_url='redis://localhost:6379/0', redis_client=None)[source]

Bases: MemoryService

A Redis-based implementation of the memory service.

Parameters:

redis_url (str)
redis_client (Redis | None)

__init__(redis_url='redis://localhost:6379/0', redis_client=None)[source]

Parameters:

redis_url (str)
redis_client (Redis | None)

async start()[source]

Starts the Redis connection.

Return type:: None

async stop()[source]

Closes the Redis connection.

Return type:: None

async health()[source]

Checks the health of the service.

Return type:: bool

async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The messages to add.
session_id (str | None) – The session id, which is optional.

Return type:

None

async search_memory(user_id, messages, filters=None)[source]

Searches messages from the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The user query or the query with history messages, both in the format of list of messages. If messages is a list, the search will be based on the content of the last message.
filters (Dict[str, Any] | None) – The filters used to search memory

Return type:

async get_query_text(message)[source]

Parameters:: message (Message)
Return type:: str

async list_memory(user_id, filters=None)[source]

Lists the memory items for a given user with filters, such as page_num, page_size, etc.

Parameters:

user_id (str) – The user id.
filters (Dict[str, Any] | None) – The filters for the memory items.

Return type:

async delete_memory(user_id, session_id=None)[source]

Deletes the memory items for a given user with certain session id, or all the memory items for a given user.

Parameters:

user_id (str)
session_id (str | None)

Return type:

None

async clear_all_memory()[source]

Clears all memory data from Redis. This method removes all user memory keys from the Redis database.

Return type:: None

async delete_user_memory(user_id)[source]

Deletes all memory data for a specific user.

Parameters:

user_id (str) – The ID of the user whose memory data should be
deleted

Return type:

None

class agentscope_runtime.engine.services.memory.tablestore_memory_service.SearchStrategy(value)[source]

Bases: Enum

An enumeration.

FULL_TEXT = 'full_text'

VECTOR = 'vector'

class agentscope_runtime.engine.services.memory.tablestore_memory_service.TablestoreMemoryService(tablestore_client, search_strategy=SearchStrategy.FULL_TEXT, embedding_model=None, vector_dimension=1536, table_name='agentscope_runtime_memory', search_index_schema=({'field_name': 'user_id', 'field_type': 4, 'index': null, 'store': null, 'is_array': null, 'enable_sort_and_agg': null, 'analyzer': null, 'analyzer_parameter': null, 'sub_field_schemas': [], 'date_formats': [], 'is_virtual_field': false, 'source_fields': [], 'vector_options': null, 'enable_highlighting': null, 'json_type': null}, {'field_name': 'session_id', 'field_type': 4, 'index': null, 'store': null, 'is_array': null, 'enable_sort_and_agg': null, 'analyzer': null, 'analyzer_parameter': null, 'sub_field_schemas': [], 'date_formats': [], 'is_virtual_field': false, 'source_fields': [], 'vector_options': null, 'enable_highlighting': null, 'json_type': null}), text_field='text', embedding_field='embedding', vector_metric_type=VectorMetricType.VM_COSINE, **kwargs)[source]

Bases: MemoryService

A Tablestore-based implementation of the memory service. based on tablestore_for_agent_memory (https://github.com/aliyun/ alibabacloud-tablestore-for-agent-memory/blob/main/python/docs/knowledge_store_tutorial.ipynb).

Parameters:

tablestore_client (AsyncOTSClient)
search_strategy (SearchStrategy)
embedding_model (Embeddings | None)
vector_dimension (int)
table_name (str | None)
search_index_schema (List[FieldSchema] | None)
text_field (str | None)
embedding_field (str | None)
vector_metric_type (VectorMetricType)
kwargs (Any)

__init__(tablestore_client, search_strategy=SearchStrategy.FULL_TEXT, embedding_model=None, vector_dimension=1536, table_name='agentscope_runtime_memory', search_index_schema=({'field_name': 'user_id', 'field_type': 4, 'index': null, 'store': null, 'is_array': null, 'enable_sort_and_agg': null, 'analyzer': null, 'analyzer_parameter': null, 'sub_field_schemas': [], 'date_formats': [], 'is_virtual_field': false, 'source_fields': [], 'vector_options': null, 'enable_highlighting': null, 'json_type': null}, {'field_name': 'session_id', 'field_type': 4, 'index': null, 'store': null, 'is_array': null, 'enable_sort_and_agg': null, 'analyzer': null, 'analyzer_parameter': null, 'sub_field_schemas': [], 'date_formats': [], 'is_virtual_field': false, 'source_fields': [], 'vector_options': null, 'enable_highlighting': null, 'json_type': null}), text_field='text', embedding_field='embedding', vector_metric_type=VectorMetricType.VM_COSINE, **kwargs)[source]

Parameters:

tablestore_client (AsyncOTSClient)
search_strategy (SearchStrategy)
embedding_model (Embeddings | None)
vector_dimension (int)
table_name (str | None)
search_index_schema (List[FieldSchema] | None)
text_field (str | None)
embedding_field (str | None)
vector_metric_type (VectorMetricType)
kwargs (Any)

async start()[source]

Start the tablestore service

Return type:: None

async stop()[source]

Close the tablestore service

Return type:: None

async health()[source]

Checks the health of the service.

Return type:: bool

async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The messages to add.
session_id (str | None) – The session id, which is optional.

Return type:

None

async static get_query_text(message)[source]

Parameters:: message (Message)
Return type:: str

async search_memory(user_id, messages, filters=None)[source]

Searches messages from the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The user query or the query with history messages, both in the format of list of messages. If messages is a list, the search will be based on the content of the last message.
filters (Dict[str, Any] | None) – The filters used to search memory

Return type:

async list_memory(user_id, filters=None)[source]

Lists the memory items for a given user with filters, such as page_num, page_size, etc.

Parameters:

user_id (str) – The user id.
filters (Dict[str, Any] | None) – The filters for the memory items.

Return type:

async delete_memory(user_id, session_id=None)[source]

Deletes the memory items for a given user with certain session id, or all the memory items for a given user.

Parameters:

user_id (str)
session_id (str | None)

Return type:

None

class agentscope_runtime.engine.services.memory.reme_personal_memory_service.ReMePersonalMemoryService(**kwargs)[source]

Bases: MemoryService

ReMe requires the following env variables to be set: FLOW_EMBEDDING_API_KEY=sk-xxxx FLOW_EMBEDDING_BASE_URL=https://xxxx/v1 FLOW_LLM_API_KEY=sk-xxxx FLOW_LLM_BASE_URL=https://xxxx/v1

__init__(**kwargs)[source]

static transform_message(message)[source]

Parameters:: message (Message)
Return type:: dict

transform_messages(messages)[source]

Parameters:: messages (List[Message])
Return type:: List[dict]

async start()[source]

Starts the service, initializing any necessary resources or connections.

Return type:: None

async stop()[source]

Stops the service, releasing any acquired resources.

Return type:: None

async health()[source]

Checks the health of the service.

Returns:: True if the service is healthy, False otherwise.
Return type:: bool

async add_memory(user_id, messages, session_id=None)[source]

Adds messages to the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The messages to add.
session_id (str | None) – The session id, which is optional.

Return type:

None

async search_memory(user_id, messages, filters=None)[source]

Searches messages from the memory service.

Parameters:

user_id (str) – The user id.
messages (list) – The user query or the query with history messages, both in the format of list of messages. If messages is a list, the search will be based on the content of the last message.
filters (Dict[str, Any] | None) – The filters used to search memory

Return type:

async list_memory(user_id, filters=None)[source]

Lists the memory items for a given user with filters, such as page_num, page_size, etc.

Parameters:

user_id (str) – The user id.
filters (Dict[str, Any] | None) – The filters for the memory items.

Return type: