ApsaraDB RDS:Chat interface usage examples

Prerequisites

Before you begin, ensure that you have:

• Activated the Professional Edition of ApsaraDB RDS Copilot

• An AccessKey ID and AccessKey secret. For security, use the AccessKey pair of a Resource Access Management (RAM) user. See View the AccessKey information of a RAM user

Set up your environment

mvn clean install

pip install -r requirements.txt

export ALIBABA_CLOUD_ACCESS_KEY_ID="<your-access-key-id>"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="<your-access-key-secret>"

Send a single question

A single-turn conversation sends one question and streams back the complete answer. The SDK method chatMessagesWithResponseIterable (Java) or chat_messages_with_sse (Python) delivers the response as a stream of Server-Sent Events (SSE). Filter for chunks where event == "message" to collect the answer text.

package com.aliyun.test;

import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.rdsai20250507.AsyncClient;
import com.aliyun.sdk.service.rdsai20250507.models.ChatMessagesRequest;
import com.aliyun.sdk.service.rdsai20250507.models.ChatMessagesResponseBody;
import darabonba.core.ResponseIterable;
import darabonba.core.ResponseIterator;
import darabonba.core.client.ClientOverrideConfiguration;
import org.apache.commons.lang3.StringUtils;

public class ChatMessagesSingleTurnDemo {
    public static void main(String[] args) {
        StaticCredentialProvider provider = StaticCredentialProvider.create(
                Credential.builder()
                        // Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
                        .accessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                        .accessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
                        //.securityToken(System.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")) // Use an STS token
                        .build());
        AsyncClient client = AsyncClient.builder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .serviceConfiguration(Configuration.create()
                        .setSignatureVersion(SignatureVersion.V3)
                ).overrideConfiguration(
                        ClientOverrideConfiguration.create()
                                .setProtocol("HTTPS")
                                // For the endpoint, see https://api.aliyun.com/product/RdsAi
                                .setEndpointOverride("rdsai.aliyuncs.com")
                ).build();

        // To use a custom agent for the conversation, pass the custom agent ID. The custom agent ID is returned after you successfully call the CreateCustomAgent operation. You can also call the ListCustomAgent operation to query the list of created custom agents. For more information, see the ListCustomAgent API documentation.
        // Inputs with a custom agent ID: ChatMessagesRequest.Inputs.builder().timezone("Asia/Shanghai").regionId("cn-hangzhou").language("zh-CN").customAgentId("5f1bbe8a-88d8-4a72-81e5-2a5d9d43****").build();
        ChatMessagesRequest.Inputs inputs = ChatMessagesRequest.Inputs.builder()
                .timezone("Asia/Shanghai")
                .regionId("cn-hangzhou")
                .language("zh-CN")
                .build();

        // For a multi-turn conversation, specify the conversationId. The conversation ID is returned after you successfully call the ChatMessages operation. For more information, see the ChatMessages API documentation.
        ChatMessagesRequest request = ChatMessagesRequest.builder()
                .inputs(inputs)
                .query("Query the list of instances in the China (Hangzhou) region")
                .build();
        try {
            ResponseIterable<ChatMessagesResponseBody> iterable = client.chatMessagesWithResponseIterable(request);
            ResponseIterator<ChatMessagesResponseBody> iterator = iterable.iterator();
            while (iterator.hasNext()) {
                ChatMessagesResponseBody chunk = iterator.next();
                if (StringUtils.equals(chunk.getEvent(), "message")) {
                    System.out.print(chunk.getAnswer());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        // Close the client
        client.close();
    }
}

# -*- coding: utf-8 -*-
import os
import sys
from typing import List

from alibabacloud_rdsai20250507.client import Client as RdsAi20250507Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_rdsai20250507 import models as rds_ai_20250507_models
from alibabacloud_tea_util import models as util_models

class ChatMessagesSingleTurnDemo:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> RdsAi20250507Client:

        # Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
        config = open_api_models.Config(
            access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        )
        # For the endpoint, see https://api.aliyun.com/product/RdsAi
        config.endpoint = 'rdsai.aliyuncs.com'
        return RdsAi20250507Client(config)

    @staticmethod
    def main() -> None:
        client = ChatMessagesSingleTurnDemo.create_client()
        # To use a custom agent for the conversation, pass the custom agent ID. The custom agent ID is returned after you successfully call the CreateCustomAgent operation. You can also call the ListCustomAgent operation to query the list of created custom agents. For more information, see the ListCustomAgent API documentation.
        # inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(language="zh-CN", region_id="cn-hangzhou", timezone="Asia/Shanghai",
        #                                                  custom_agent_id="5f1bbe8a-88d8-4a72-81e5-2a5d9d43****")
        inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(
            language="zh-CN",
            region_id="cn-hangzhou",
            timezone="Asia/Shanghai"
        )
        # For a multi-turn conversation, specify the conversationId. The conversation ID is returned after you successfully call the ChatMessages operation. For more information, see the ChatMessages API documentation.
        chat_messages_request = rds_ai_20250507_models.ChatMessagesRequest(
            query="Query the list of instances in the China (Hangzhou) region",
            inputs=inputs
        )

        runtime = util_models.RuntimeOptions()
        chat_messages_response = client.chat_messages_with_sse(tmp_req=chat_messages_request, runtime=runtime)
        for chunk in chat_messages_response:
            body = chunk.body
            if body is not None and body.event == 'message':
                print(f"{body.answer}", end="")

if __name__ == '__main__':
    ChatMessagesSingleTurnDemo.main()

Maintain a multi-turn conversation

Multi-turn conversations carry context across messages using a conversationId. The first turn works exactly like a single-turn request — no conversationId is required. The response includes a conversationId; pass it in every subsequent request to continue the same session. Optionally, call GetMessages with the conversationId to retrieve the full conversation history.

package com.aliyun.test;

import com.alibaba.fastjson.JSON;
import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.rdsai20250507.AsyncClient;
import com.aliyun.sdk.service.rdsai20250507.models.*;
import darabonba.core.ResponseIterable;
import darabonba.core.client.ClientOverrideConfiguration;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;

public class ChatMessagesMultiTurnDemo {
    public interface AnswerCallback {
        void onAnswerChunk(ChatMessagesResponseBody chunk);
    }

    private AsyncClient createClient() throws Exception {
        StaticCredentialProvider provider = StaticCredentialProvider.create(
                Credential.builder()
                        // Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
                        .accessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                        .accessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
                        //.securityToken(System.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")) // Use an STS token
                        .build());
        AsyncClient client = AsyncClient.builder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .serviceConfiguration(Configuration.create()
                        .setSignatureVersion(SignatureVersion.V3)
                ).overrideConfiguration(
                        ClientOverrideConfiguration.create()
                                .setProtocol("HTTPS")
                                // For the endpoint, see https://api.aliyun.com/product/RdsAi
                                .setEndpointOverride("rdsai.aliyuncs.com")
                ).build();
        return client;
    }

    /**
     * Sends a conversation message
     */
    public void chatMessages(String query, String conversationId, ChatMessagesRequest.Inputs inputs, ChatMessagesMultiTurnDemo.AnswerCallback callback) throws Exception {
        try (AsyncClient client = createClient()) {
            ChatMessagesRequest request = ChatMessagesRequest.builder()
                    .conversationId(conversationId)
                    .inputs(inputs)
                    .query(query)
                    .build();
            ;
            ResponseIterable<ChatMessagesResponseBody> iterable = client.chatMessagesWithResponseIterable(request);
            for (ChatMessagesResponseBody event : iterable) {
                callback.onAnswerChunk(event);
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /**
     * Views a specific conversation message
     */
    public GetMessagesResponseBody getMessages(String conversationId) {
        try (AsyncClient client = createClient()) {
            GetMessagesRequest request = GetMessagesRequest.builder()
                    .conversationId(conversationId)
                    .build();
            CompletableFuture<GetMessagesResponse> messages = client.getMessages(request);
            return messages.get().getBody();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        ChatMessagesMultiTurnDemo messagesMultiTurnDemo = new ChatMessagesMultiTurnDemo();
        final AtomicReference<String> conversationId = new AtomicReference<>();

        ChatMessagesRequest.Inputs inputs = ChatMessagesRequest.Inputs.builder()
                .timezone("Asia/Shanghai")
                .regionId("cn-hangzhou")
                .language("zh-CN")
                .build();
        // Get the conversation ID from the first turn
        messagesMultiTurnDemo.chatMessages("Query the list of instances in the China (Hangzhou) region", null, inputs, new ChatMessagesMultiTurnDemo.AnswerCallback() {
            @Override
            public void onAnswerChunk(ChatMessagesResponseBody event) {
                if ("message".equals(event.getEvent())) {
                    String answer = event.getAnswer();
                    if (answer != null) {
                        System.out.print(answer);
                    }
                    conversationId.set(event.getConversationId());
                }
            }
        });
        System.out.println(System.lineSeparator() + "conversationId:" + conversationId.get());
        // Start the second turn using the obtained conversation ID
        messagesMultiTurnDemo.chatMessages("Query the details of the first instance", conversationId.get(), inputs, new ChatMessagesMultiTurnDemo.AnswerCallback() {
            @Override
            public void onAnswerChunk(ChatMessagesResponseBody event) {
                if ("message".equals(event.getEvent())) {
                    String answer = event.getAnswer();
                    if (answer != null) {
                        System.out.print(answer);
                    }
                }
            }
        });

        GetMessagesResponseBody messagesResponseBody = messagesMultiTurnDemo.getMessages(conversationId.get());
        System.out.println(System.lineSeparator() + "messagesResponseBody:" + JSON.toJSONString(messagesResponseBody));

    }
}

# -*- coding: utf-8 -*-
import os
import sys
from typing import List

from alibabacloud_rdsai20250507.client import Client as RdsAi20250507Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_rdsai20250507 import models as rds_ai_20250507_models
from alibabacloud_tea_util import models as util_models


class ChatMessagesMultiTurnDemo:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> RdsAi20250507Client:

        # Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
        config = open_api_models.Config(
            access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        )
        # For the endpoint, see https://api.aliyun.com/product/RdsAi
        config.endpoint = 'rdsai.aliyuncs.com'
        return RdsAi20250507Client(config)

    # Sends a conversation message
    def do_chatMessages(self, search_query: str, conversation_id: str):
        runtime = util_models.RuntimeOptions(read_timeout=1000 * 100)
        # To use a custom agent for the conversation, pass the custom agent ID. The custom agent ID is returned after you successfully call the CreateCustomAgent operation. You can also call the ListCustomAgent operation to query the list of created custom agents. For more information, see the ListCustomAgent API documentation.
        # inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(language="zh-CN", region_id="cn-hangzhou", timezone="Asia/Shanghai",
        #                                                  custom_agent_id="5f1bbe8a-88d8-4a72-81e5-2a5d9d43****")
        inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(
            language="zh-CN",
            region_id="cn-hangzhou",
            timezone="Asia/Shanghai"
        )
        # For a multi-turn conversation, specify the conversationId. The conversation ID is returned after you successfully call the ChatMessages operation. For more information, see the ChatMessages API documentation.
        request = rds_ai_20250507_models.ChatMessagesRequest(
            query=search_query,
            inputs=inputs,
            conversation_id=conversation_id
        )
        sse_response = self.create_client().chat_messages_with_sse(tmp_req=request, runtime=runtime)
        return sse_response

    # Views a specific conversation message
    def do_getMessages(self, conversation_id: str):
        request = rds_ai_20250507_models.GetMessagesRequest(conversation_id=conversation_id)
        messages = self.create_client().get_messages(request)
        if messages and messages.body:
            return messages.body
        return None

    @staticmethod
    def main() -> None:
        chat_message = ChatMessagesMultiTurnDemo()
        conversation_id = None

        # Get the conversation ID from the first turn
        chat_messages_response = chat_message.do_chatMessages(search_query="Query the list of instances in the China (Hangzhou) region", conversation_id=None)
        for chunk in chat_messages_response:
            body = chunk.body
            if body is not None and body.event == 'message':
                print(f"{body.answer}", end="")
                conversation_id = body.conversation_id

        assert conversation_id is not None

        # Start the second turn using the obtained conversation ID
        chat_messages_response = chat_message.do_chatMessages(search_query="Query the details of the first instance", conversation_id=conversation_id)
        for chunk in chat_messages_response:
            body = chunk.body
            if body is not None and body.event == 'message':
                print(f"{body.answer}", end="")

        messages_body = chat_message.do_getMessages(conversation_id)
        print("\n" + str(messages_body))


if __name__ == '__main__':
    ChatMessagesMultiTurnDemo.main()

Create and use a custom agent

A custom agent is an AI assistant with a specific role, system prompt, and a set of tools. Create one with CreateCustomAgent, which returns a customAgentId. Pass that ID in the inputs parameter of ChatMessages to activate the agent's behavior for all subsequent turns.

After creation, manage agents with these operations:

package com.aliyun.test;

import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.rdsai20250507.AsyncClient;
import com.aliyun.sdk.service.rdsai20250507.models.*;
import darabonba.core.ResponseIterable;
import darabonba.core.client.ClientOverrideConfiguration;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class CustomAgentDemo {
    public interface AnswerCallback {
        void onAnswerChunk(ChatMessagesResponseBody chunk);
    }

    private AsyncClient createClient() throws Exception {
        StaticCredentialProvider provider = StaticCredentialProvider.create(
                Credential.builder()
                        // Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
                        .accessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                        .accessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
                        //.securityToken(System.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")) // Use an STS token
                        .build());
        AsyncClient client = AsyncClient.builder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .serviceConfiguration(Configuration.create()
                        .setSignatureVersion(SignatureVersion.V3)
                ).overrideConfiguration(
                        ClientOverrideConfiguration.create()
                                .setProtocol("HTTPS")
                                // For the endpoint, see https://api.aliyun.com/product/RdsAi
                                .setEndpointOverride("rdsai.aliyuncs.com")
                ).build();
        return client;
    }

    /**
     * Sends a conversation message
     */
    public void chatMessages(String query, String conversationId, ChatMessagesRequest.Inputs inputs, CustomAgentDemo.AnswerCallback callback) throws Exception {
        try (AsyncClient client = createClient()) {
            ChatMessagesRequest request = ChatMessagesRequest.builder()
                    .conversationId(conversationId)
                    .inputs(inputs)
                    .query(query)
                    .build();
            ;
            ResponseIterable<ChatMessagesResponseBody> iterable = client.chatMessagesWithResponseIterable(request);
            for (ChatMessagesResponseBody event : iterable) {
                callback.onAnswerChunk(event);
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /**
     * Creates a custom agent
     */
    public String createCustomAgent(String agentName, String prompt, boolean enableTools, List<String> tools) throws Exception {
        CompletableFuture<CreateCustomAgentResponse> response;
        try (AsyncClient client = createClient()) {
            CreateCustomAgentRequest request = CreateCustomAgentRequest.builder()
                    .name(agentName)
                    .systemPrompt(prompt)
                    .enableTools(enableTools)
                    .tools(tools)
                    .build();
            response = client.createCustomAgent(request);
            return response.get().getBody().getId();
        }
    }

    /**
     * Gets a custom agent by name
     */
    public ListCustomAgentResponseBody.Data getCustomAgent(String agentName) throws Exception {
        CompletableFuture<ListCustomAgentResponse> response;
        try (AsyncClient client = createClient()) {
            ListCustomAgentRequest request = ListCustomAgentRequest.builder()
                    .pageNumber(1L)
                    .pageSize(30L)
                    .build();
            response = client.listCustomAgent(request);
            for (ListCustomAgentResponseBody.Data agent : response.get().getBody().getData()) {
                if (agent.getName().equals(agentName)) {
                    return agent;
                }
            }
        }
        return null;
    }

    /**
     * Updates a custom agent
     */
    public String updateCustomAgent(String agentId, String prompt, boolean enableTools, List<String> tools) throws Exception {
        CompletableFuture<UpdateCustomAgentResponse> response;
        try (AsyncClient client = createClient()) {
            UpdateCustomAgentRequest request = UpdateCustomAgentRequest.builder()
                    .customAgentId(agentId)
                    .systemPrompt(prompt)
                    .enableTools(enableTools)
                    .tools(tools)
                    .build();
            response = client.updateCustomAgent(request);
            return response.get().getBody().getId();
        }
    }

    /**
     * Deletes a custom agent
     */
    public String deleteCustomAgent(String agentId) throws Exception {
        CompletableFuture<DeleteCustomAgentResponse> response;
        try (AsyncClient client = createClient()) {
            DeleteCustomAgentRequest request = DeleteCustomAgentRequest.builder()
                    .customAgentId(agentId)
                    .build();
            response = client.deleteCustomAgent(request);
            return response.get().getBody().getResult();
        }
    }

    public static void main(String[] args) throws Exception {
        CustomAgentDemo messagesMultiTurnDemo = new CustomAgentDemo();

        String agentName = "copilot-agent";
        // Create a user agent
        String customAgentId = messagesMultiTurnDemo.createCustomAgent(agentName, "copilot-agent-prompt", true, Arrays.asList("describe_db_instances"));

        // Update the user agent
        String updateCustomAgentId = messagesMultiTurnDemo.updateCustomAgent(customAgentId, "You are a custom agent specialized in retrieving a list of instance information for a specified region. You can call the [describe_db_instances] tool to help you get the information and generate a detailed summary for me.", true, Arrays.asList("describe_db_instances"));

        // Query the user agent
        ListCustomAgentResponseBody.Data customAgent = messagesMultiTurnDemo.getCustomAgent(agentName);
        if (!customAgent.getName().equals(agentName)) {
            throw new RuntimeException("Agent name mismatch: expected " + agentName + ", got " + customAgent.getName());
        }
        if (!customAgent.getSystemPrompt().equals("You are a custom agent specialized in retrieving a list of instance information for a specified region. You can call the [describe_db_instances] tool to help you get the information and generate a detailed summary for me.")) {
            throw new RuntimeException("Agent prompt mismatch: expected 'You are a custom agent specialized in retrieving a list of instance information for a specified region. You can call the [describe_db_instances] tool to help you get the information and generate a detailed summary for me.', got " + customAgent.getSystemPrompt());
        }

        ChatMessagesRequest.Inputs inputs = ChatMessagesRequest.Inputs.builder()
                .timezone("Asia/Shanghai")
                .regionId("cn-hangzhou")
                .language("zh-CN")
                .customAgentId(customAgentId)
                .build();

        // Start the conversation
        messagesMultiTurnDemo.chatMessages("Query which instances in the China (Hangzhou) region are about to expire", null, inputs, new CustomAgentDemo.AnswerCallback() {
            @Override
            public void onAnswerChunk(ChatMessagesResponseBody event) {
                if ("message".equals(event.getEvent())) {
                    String answer = event.getAnswer();
                    if (answer != null) {
                        System.out.print(answer);
                    }
                }
            }
        });

        //Delete the user agent
        String deleteCustomAgent = messagesMultiTurnDemo.deleteCustomAgent(customAgentId);
        System.out.println(System.lineSeparator() + deleteCustomAgent);
    }
}

# -*- coding: utf-8 -*-
import os

from alibabacloud_rdsai20250507.client import Client as RdsAi20250507Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_rdsai20250507 import models as rds_ai_20250507_models
from alibabacloud_tea_util import models as util_models

class CustomAgentDemo:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> RdsAi20250507Client:

        # Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set.
        config = open_api_models.Config(
            access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        )
        # For the endpoint, see https://api.aliyun.com/product/RdsAi
        config.endpoint = 'rdsai.aliyuncs.com'
        return RdsAi20250507Client(config)

    # Sends a conversation message
    def do_chat_messages(self, search_query: str, custom_agent_id: str, conversation_id: str):
        runtime = util_models.RuntimeOptions(read_timeout=1000 * 100)
        # To use a custom agent for the conversation, pass the custom agent ID. The custom agent ID is returned after you successfully call the CreateCustomAgent operation. You can also call the ListCustomAgent operation to query the list of created custom agents. For more information, see the ListCustomAgent API documentation.
        # inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(language="zh-CN", region_id="cn-hangzhou", timezone="Asia/Shanghai",
        #                                                  custom_agent_id="5f1bbe8a-88d8-4a72-81e5-2a5d9d43****")
        inputs = rds_ai_20250507_models.ChatMessagesRequestInputs(
            language="zh-CN",
            region_id="cn-hangzhou",
            timezone="Asia/Shanghai",
            custom_agent_id=custom_agent_id
        )
        # For a multi-turn conversation, specify the conversationId. The conversation ID is returned after you successfully call the ChatMessages operation. For more information, see the ChatMessages API documentation.
        request = rds_ai_20250507_models.ChatMessagesRequest(
            query=search_query,
            inputs=inputs,
            conversation_id=conversation_id
        )
        sse_response = self.create_client().chat_messages_with_sse(tmp_req=request, runtime=runtime)
        return sse_response

    # Creates a custom agent
    def do_create_custom_agent(self, agent_name: str, prompt: str,
                               enable_tools: bool, tools: list) -> str:
        try:
            req = rds_ai_20250507_models.CreateCustomAgentRequest(name=agent_name, system_prompt=prompt, enable_tools=enable_tools, tools=tools)
            resp = self.create_client().create_custom_agent(req)
            return resp.body.id
        except Exception as e:
            raise RuntimeError(e)

    # Updates a custom agent
    def do_update_custom_agent(self, custom_agent_id: str, system_prompt: str) -> rds_ai_20250507_models.UpdateCustomAgentResponseBody:
        try:
            req = rds_ai_20250507_models.UpdateCustomAgentRequest(custom_agent_id=custom_agent_id, system_prompt=system_prompt)
            resp = self.create_client().update_custom_agent(req)
            return resp.body
        except Exception as e:
            raise RuntimeError(e)

    def do_get_custom_agent(self, agent_name: str, page_number: int, page_size: int):
        try:
            req = rds_ai_20250507_models.ListCustomAgentRequest(page_number=page_number, page_size=page_size)
            resp = self.create_client().list_custom_agent(req)
            body = resp.body
            if not body:
                return None
            data_list = body.data
            if not data_list:
                return None
            for data in data_list:
                if getattr(data, "name", None) == agent_name:
                    return getattr(data, "id", None)
        except Exception as e:
            raise RuntimeError(e)
        return None

    # Deletes a custom agent
    def do_delete_custom_agent(self, custom_agent_id: str):
        try:
            req = rds_ai_20250507_models.DeleteCustomAgentRequest(custom_agent_id=custom_agent_id)
            resp = self.create_client().delete_custom_agent(req)
            return resp.body
        except Exception as e:
            raise RuntimeError(e)

    @staticmethod
    def main() -> None:
        custom_agent = CustomAgentDemo()
        agent_name = "copilot-agent-python"
        create_custom_agent_id = custom_agent.do_create_custom_agent(agent_name=agent_name, prompt="copilot-agent-prompt-demo", enable_tools=True,
                                                                     tools=["describe_db_instances"])
        print("custom_agent_id: " + create_custom_agent_id)

        update_agent_body = custom_agent.do_update_custom_agent(custom_agent_id=create_custom_agent_id, system_prompt="You are a custom agent specialized in retrieving a list of instance information for a specified region. You can call the [describe_db_instances] tool to help you get the information and generate a detailed summary for me.")
        print("custom_agent_id_update: " + update_agent_body.id)

        custom_agent_id_by_query = custom_agent.do_get_custom_agent(agent_name, 1, 30)
        print("custom_agent_id: " + custom_agent_id_by_query)

        chat_messages_response = custom_agent.do_chat_messages(search_query="Query which instances in the China (Hangzhou) region are about to expire", custom_agent_id=create_custom_agent_id,
                                                               conversation_id=None)
        for chunk in chat_messages_response:
            body = chunk.body
            if body is not None and body.event == 'message':
                print(f"{body.answer}", end="")
                conversation_id = body.conversation_id

        del_custom_agent_body = custom_agent.do_delete_custom_agent(custom_agent_id=create_custom_agent_id)
        print("\n" + "delete_agent_result: " + del_custom_agent_body.result)

if __name__ == '__main__':
    CustomAgentDemo.main()

Download sample code

Download complete, runnable sample code covering all three scenarios:

RDS_Copilot_Examples.zip

API reference

• ChatMessages - Send a conversation message

• GetMessages - View conversation messages

• CreateCustomAgent - Create a custom agent

• ListCustomAgent - View a list of custom agents

• UpdateCustomAgent - Update a custom agent

• DeleteCustomAgent - Delete a custom agent