Chat to deploy and manage applications on any infrastructure
"""Appilot agent."""

agent/__init__.py
seal-io/appilot
from typing import Any, Dict, Optional

from langchain.tools import BaseTool
from langchain.agents.agent import AgentExecutor
from langchain.agents.conversational.base import ConversationalAgent
from langchain.callbacks.base import BaseCallbackManager
from langchain.chains.llm import LLMChain
from langchain.memory import ReadOnlySharedMemory
from langchain.schema.language_model import BaseLanguageModel

from config import config
from tools.human.tool import HumanTool
from tools.reasoning.tool import ShowReasoningTool, HideReasoningTool
from agent.output_parser import OutputParser
from agent.prompt import (
    AGENT_PROMPT_PREFIX,
    FORMAT_INSTRUCTIONS_TEMPLATE,
)


def create_agent(
    llm: BaseLanguageModel,
    shared_memory: Optional[ReadOnlySharedMemory] = None,
    tools: list[BaseTool] = [],
    callback_manager: Optional[BaseCallbackManager] = None,
    verbose: bool = True,
    agent_executor_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs: Dict[str, Any],
) -> AgentExecutor:
    """Instantiate planner for a given task."""

    system_tools = [
        HumanTool(),
        ShowReasoningTool(),
        HideReasoningTool(),
    ]

    tools.extend(system_tools)

    format_instructions = FORMAT_INSTRUCTIONS_TEMPLATE.format(
        natural_language=config.APPILOT_CONFIG.natural_language
    )
    prompt = ConversationalAgent.create_prompt(
        tools,
        prefix=AGENT_PROMPT_PREFIX,
        format_instructions=format_instructions,
    )

    agent = ConversationalAgent(
        llm_chain=LLMChain(
            llm=llm, prompt=prompt, verbose=config.APPILOT_CONFIG.verbose
        ),
        output_parser=OutputParser(),
        allowed_tools=[tool.name for tool in tools],
        **kwargs,
    )

    return AgentExecutor.from_agent_and_tools(
        agent=agent,
        tools=tools,
        memory=shared_memory,
        callback_manager=callback_manager,
        verbose=verbose,
        **(agent_executor_kwargs or {}),
    )

agent/agent.py
seal-io/appilot
import re
from typing import Union

from langchain.agents.agent import AgentOutputParser
from langchain.agents.conversational.prompt import FORMAT_INSTRUCTIONS
from langchain.schema import AgentAction, AgentFinish, OutputParserException


class OutputParser(AgentOutputParser):
    """Output parser for the agent. It extends the convoAgent parser to support multiline observation."""

    ai_prefix: str = "AI"
    """Prefix to use before AI output."""

    def get_format_instructions(self) -> str:
        return FORMAT_INSTRUCTIONS

    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        if f"{self.ai_prefix}:" in text:
            return AgentFinish(
                {"output": text.split(f"{self.ai_prefix}:")[-1].strip()}, text
            )
        regex = r"Action: (.*?)[\n]*Action Input: (.*?)\nReason: .*"
        match = re.search(regex, text, re.DOTALL)
        if not match:
            raise OutputParserException(
                f"Could not parse LLM output: `{text}`"
            )
        action = match.group(1)
        action_input = match.group(2)
        return AgentAction(
            action.strip(), action_input.strip(" ").strip('"'), text
        )

    @property
    def _type(self) -> str:
        return "conversational"

agent/output_parser.py
seal-io/appilot
AGENT_PROMPT_PREFIX = """You are Appilot, an agent that assists with user queries for application management.
Only take action to user queries that are relevant to application management.
Example:
```
User query: tell me a joke
Plan: Sorry, I specialize in application management, not comedy.
```

Notes:
ID of service is a string that looks like a long number.
Always use the construct_service tool before creating or updating a service.

TOOLS:
------

You have access to the following tools:"""

FORMAT_INSTRUCTIONS_TEMPLATE = """To use a tool, please use the following format:

```
Thought: Do I need to use a tool? Yes
Action: the action to take, should be one of [{{tool_names}}]
Action Input: the input to the action
Reason: the reason you use this tool(in {natural_language})
Observation: the result of the action
```

When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:

```
Thought: Do I need to use a tool? No
Reason: the reason you do not need to use a tool
{{ai_prefix}}: [your response here]
```

In the reponse, don't show project id and environment id. Unless user explicitly ask for it.
Use markdown format for the response. If the data is suitable to show in table, use markdown table.
Please print the response to human in {natural_language}.
"""

agent/prompt.py
seal-io/appilot
from command import cli


cli.run()

app.py
seal-io/appilot
import json
from typing import Any, Dict, Optional, List
from uuid import UUID
import click
import yaml

from i18n import text
from utils import utils
from config import config

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema.output import LLMResult
from langchain.schema.output import LLMResult
from pygments import highlight
from pygments.lexers import JsonLexer
from pygments.lexers import YamlLexer
from pygments.formatters import TerminalFormatter


class HumanRejectedException(Exception):
    """Exception to raise when a person manually review and rejects a value."""


def remove_triple_backticks(text):
    if text.startswith("```") and text.endswith("```"):
        lines = text.split("\n")

        if len(lines) <= 1:
            return ""

        lines = lines[1:-1]

        result_text = "\n".join(lines)

        return result_text

    return text


class ApprovalCallbackHandler(BaseCallbackHandler):
    """Callback for manual approval."""

    raise_error: bool = True

    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        **kwargs: Any,
    ) -> Any:
        if not self._approve(input_str, serialized):
            raise HumanRejectedException(
                f"Inputs {input_str} to tool {serialized} were rejected."
            )

    def _approve(self, _input: str, serialized: Dict[str, Any]) -> bool:
        message = text.get("ask_approval")
        _input = remove_triple_backticks(_input.strip())

        is_json = False
        try:
            json_input = json.loads(_input)
        except Exception as e:
            # If the input is not a valid JSON, just pass
            pass
        else:
            # Serialize the JSON input with colors
            is_json = True
            json_string = json.dumps(json_input, indent=4)
            _input = highlight(json_string, JsonLexer(), TerminalFormatter())

        if not is_json:
            # Now try YAML
            try:
                yaml.safe_load_all(_input)
            except Exception as e:
                # If the input is not a valid YAML, just pass
                pass
            else:
                # Serialize the YAML input with colors
                _input = highlight(_input, YamlLexer(), TerminalFormatter())

        return click.confirm(
            message.format(input=_input, tool_name=serialized["name"]),
            default=False,
            prompt_suffix="",
        )


class PrintReasoningCallbackHandler(BaseCallbackHandler):
    def on_llm_end(
        self,
        response: LLMResult,
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> None:
        if not config.APPILOT_CONFIG.show_reasoning:
            return

        generation_text = response.generations[0][0].text
        lines = generation_text.splitlines()
        reason_prompt_prefix = "Reason:"
        for line in reversed(lines):
            if line.startswith(reason_prompt_prefix):
                reason_text = line.lstrip("Reason:").strip()
                utils.print_ai_reasoning(reason_text)
                break
        """Print AI reasoning."""

callbacks/handlers.py
seal-io/appilot
"""Appilot commands."""

command/__init__.py
seal-io/appilot
import sys
from typing import Any
import readline


from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
import colorama

from callbacks import handlers
from config import config
from i18n import text
from utils import utils
from agent.agent import create_agent
from walrus.toolkit import WalrusToolKit
from k8s.toolkit import KubernetesToolKit

last_error = None


def setup_agent() -> Any:
    config.init()
    colorama.init()

    llm = ChatOpenAI(
        model_name="gpt-4",
        temperature=0,
        callbacks=[handlers.PrintReasoningCallbackHandler()],
    )

    text.init_system_messages(llm)

    memory = ConversationBufferMemory(memory_key="chat_history")

    enabled_toolkits = [
        toolkit.lower() for toolkit in config.APPILOT_CONFIG.toolkits
    ]

    tools = []
    if "kubernetes" in enabled_toolkits:
        kubernetes_toolkit = KubernetesToolKit(llm=llm)
        tools.extend(kubernetes_toolkit.get_tools())
    elif "walrus" in enabled_toolkits:
        walrus_toolkit = WalrusToolKit(llm=llm)
        tools.extend(walrus_toolkit.get_tools())
    else:
        print(text.get("enable_no_toolkit"))
        sys.exit(1)

    return create_agent(
        llm,
        shared_memory=memory,
        tools=tools,
        verbose=config.APPILOT_CONFIG.verbose,
    )


def run():
    appilot_agent = setup_agent()

    print(text.get("welcome"))
    user_query = None
    while True:
        user_query = input(">")
        if utils.is_inform_sent():
            continue
        elif user_query == "exit":
            break
        elif user_query == "appilot_log":
            print_last_error()
            continue
        elif user_query.startswith("#"):
            continue
        elif not user_query.strip():
            continue

        try:
            result = appilot_agent.run(user_query)
        except handlers.HumanRejectedException as he:
            utils.print_rejected_message()
            continue
        except Exception as e:
            handle_exception(e)
            continue

        utils.print_ai_response(result)


def handle_exception(e):
    global last_error
    print(text.get("response_prefix"), end="")
    print(text.get("error_occur_message"))
    last_error = e


def print_last_error():
    if last_error is None:
        print(text.get("response_prefix"), end="")
        print(text.get("no_error_message"))
    else:
        print(last_error)

command/cli.py
seal-io/appilot
"""Global configuration of Appilot."""

config/__init__.py
seal-io/appilot
import logging

import urllib3
from utils import utils

from pydantic import BaseModel
from dotenv import load_dotenv


class Config(BaseModel):
    openai_api_key: str
    openai_api_base: str
    natural_language: str
    toolkits: list[str]
    show_reasoning: bool
    verbose: bool


APPILOT_CONFIG: Config


def init():
    load_dotenv()
    openai_api_base = utils.get_env("OPENAI_API_BASE")
    openai_api_key = utils.get_env("OPENAI_API_KEY")
    natural_language = utils.get_env("NATURAL_LANGUAGE", "English")
    toolkits = utils.get_env_list("TOOLKITS")
    show_reasoning = utils.get_env_bool("SHOW_REASONING", True)
    verbose = utils.get_env_bool("VERBOSE", False)

    if not openai_api_key:
        raise Exception("OPENAI_API_KEY is not set")

    if not verbose:
        logging.basicConfig(level=logging.CRITICAL)
        # Disable child loggers of urllib3, e.g. urllib3.connectionpool
        logging.getLogger(urllib3.__package__).propagate = False

    global APPILOT_CONFIG
    APPILOT_CONFIG = Config(
        openai_api_base=openai_api_base,
        openai_api_key=openai_api_key,
        natural_language=natural_language,
        toolkits=toolkits,
        show_reasoning=show_reasoning,
        verbose=verbose,
    )


def set_verbose(verbose: bool):
    global APPILOT_CONFIG
    APPILOT_CONFIG.verbose = verbose


def set_show_reasoning(show_reasoning: bool):
    global APPILOT_CONFIG
    APPILOT_CONFIG.show_reasoning = show_reasoning

config/config.py
seal-io/appilot
"""For internationalization. Most of the job is done by prompts. For the rest system messages, we also use AI to translate them."""

i18n/__init__.py
seal-io/appilot
import json

from config import config

from langchain.schema.language_model import BaseLanguageModel

prompt = """
Translate the following json map to {language}. Keep the keys unchanged.
{messages}

RESULT:

"""

system_messages = {
    "welcome": "Appilot: What can I help?",
    "ai_reasoning": "Appilot reasoning: ",
    "response_prefix": "Appilot: ",
    "inform_prefix": "Appilot[inform]: ",
    "error_occur_message": "An internal error occurred. Enter 'appilot_log' if you want to see the details.",
    "rejected_message": "The action is rejected.",
    "no_error_message": "No error occurred.",
    "resource_log_prefix": "Here's the log:",
    "watch_service_note": "( Enter <Ctrl + C> to halt )",
    "watch_service_ending": "Halted.",
    "show_graph_message": "The dependency graph is shown to you.",
    "inform_ready_start": "Start watching. Will inform when it's ready.",
    "service_ready_message": "Service {} is Ready.",
    "enable_no_toolkit": "No toolkit available. Please enable at least one toolkit.",
    "ask_approval": """
The following action requires approval:

Input:
{input}

Action: 
{tool_name}
 
Do you approve the above action? """,
}


def init_system_messages(llm: BaseLanguageModel):
    language = config.APPILOT_CONFIG.natural_language
    if language.lower() in ("en", "english"):
        return

    global system_messages
    system_messages_string = json.dumps(system_messages, ensure_ascii=False)
    result = llm.predict(
        prompt.format(language=language, messages=system_messages_string)
    )
    translated = json.loads(result)
    system_messages = translated


def get(key):
    return system_messages[key]

i18n/text.py
seal-io/appilot
from typing import Any
from kubernetes import client


API_RESOURCES: list[dict[Any | str, Any]]


class GroupVersionKind:
    """GroupVersionKind."""

    def __init__(self, groupVersion: str, kind: str):
        self.groupVersion = groupVersion
        self.kind = kind


def init_api_resources_cache():
    """Get available api resources. Similar to kubectl api-resources."""

    api_client = client.ApiClient()
    api_resource_list = api_client.call_api(
        "/api/v1",
        "GET",
        response_type="object",
        _return_http_data_only=True,
    )
    api_resources = [
        {**resource, "groupVersion": "v1"}
        for resource in api_resource_list["resources"]
        if "storageVersionHash" in resource
    ]

    api_group_list = api_client.call_api(
        "/apis",
        "GET",
        response_type="object",
        _return_http_data_only=True,
    )

    for api_group in api_group_list["groups"]:
        api_resource_list = api_client.call_api(
            "/apis/" + api_group["preferredVersion"]["groupVersion"],
            "GET",
            response_type="object",
            _return_http_data_only=True,
        )
        api_resources.extend(
            [
                {
                    **resource,
                    "groupVersion": api_group["preferredVersion"][
                        "groupVersion"
                    ],
                }
                for resource in api_resource_list["resources"]
                if "storageVersionHash" in resource and resource["namespaced"]
            ]
        )
    global API_RESOURCES
    API_RESOURCES = api_resources


def get_api_resources():
    return API_RESOURCES


def search_api_resource(resource_kind: str) -> GroupVersionKind:
    api_resources = get_api_resources()
    matching_resources = [
        api_resource
        for api_resource in api_resources
        if str(api_resource["name"]).lower() == resource_kind
        or str(api_resource["singularName"]).lower() == resource_kind
        or str(api_resource["kind"]).lower() == resource_kind
        or (
            "shortNames" in api_resource
            and resource_kind in api_resource["shortNames"]
        )
    ]

    if matching_resources:
        return GroupVersionKind(
            matching_resources[0]["groupVersion"],
            matching_resources[0]["kind"],
        )

    raise Exception(f"Resource {resource_kind} not found.")

k8s/context.py
seal-io/appilot
import logging
import subprocess
import sys
from langchain.schema.language_model import BaseLanguageModel
from kubernetes import config, client

from k8s import context
from k8s.tools.helm.tool import (
    DeleteApplicationTool,
    DeployApplicationTool,
    GenerateUpgradeApplicationValuesTool,
    GetApplicationAccessEndpointsTool,
    GetApplicationDetailTool,
    ListApplicationsTool,
    SearchChartTool,
    UpgradeApplicationTool,
)
from k8s.tools.manage_resource.tool import (
    ApplyResourcesTool,
    ConstructResourceForUpdateTool,
    ConstructResourceTool,
    DeleteResourceTool,
    DescribePodTool,
    GetIngressAccessEndpointsTool,
    GetPodLogsTool,
    GetResourceDetailTool,
    GetResourceYamlTool,
    GetServiceAccessEndpointsTool,
    ListResourcesForInfoTool,
    ListResourcesTool,
    WatchResourcesTool,
)
from walrus.tools.general.tools import BrowseURLTool

logger = logging.getLogger(__name__)


def command_installed(commands: list[str]):
    try:
        result = subprocess.run(
            commands,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )

        if result.returncode == 0:
            return True
        else:
            return False
    except FileNotFoundError:
        return False


class KubernetesToolKit:
    """Kubernetes toolkit."""

    llm: BaseLanguageModel

    def __init__(self, llm: BaseLanguageModel):
        self.llm = llm
        config.load_kube_config()
        self.precheck()
        context.init_api_resources_cache()

    def precheck(self):
        if not command_installed(["kubectl", "version", "--client"]):
            print("Precheck failed: kubectl not found.")
            sys.exit(1)

        if not command_installed(["helm", "version"]):
            print("Precheck failed: helm not found.")
            sys.exit(1)

        try:
            client.VersionApi().get_code(_request_timeout=2)
        except Exception as e:
            logger.debug("Error connecting to Kubernetes cluster: {e}")
            print("Precheck failed: Kubernetes cluster is not available.")
            sys.exit(1)

    def get_tools(self):
        llm = self.llm
        tools = [
            ListResourcesTool(return_direct=True),
            ListResourcesForInfoTool(),
            GetResourceDetailTool(),
            GetResourceYamlTool(return_direct=True),
            GetServiceAccessEndpointsTool(),
            GetIngressAccessEndpointsTool(),
            GetPodLogsTool(),
            DescribePodTool(),
            WatchResourcesTool(return_direct=True),
            DeleteResourceTool(),
            ConstructResourceTool(llm=llm),
            ConstructResourceForUpdateTool(llm=llm),
            ApplyResourcesTool(),
            SearchChartTool(llm=llm),
            DeployApplicationTool(),
            GenerateUpgradeApplicationValuesTool(llm=llm),
            UpgradeApplicationTool(),
            ListApplicationsTool(),
            GetApplicationDetailTool(),
            GetApplicationAccessEndpointsTool(),
            BrowseURLTool(),
            DeleteApplicationTool(),
        ]
        return tools

k8s/toolkit.py
seal-io/appilot
from kubernetes import config, client


def get_service_endpoints(service):
    service_endpoints = []
    if service.spec.type == "NodePort":
        service_endpoints.extend(get_nodeport_service_endpoints(service))
    elif service.spec.type == "LoadBalancer":
        service_endpoints.extend(get_loadbalancer_service_endpoints(service))

    return service_endpoints


def get_loadbalancer_service_endpoints(service):
    if (
        service.status.load_balancer
        and "ingress" in service.status.load_balancer
    ):
        endpoints = []
        for ingress in service.status.load_balancer.ingress:
            if ingress.hostname != "":
                endpoints.append(
                    {
                        "name": service.name,
                        "endpoint": ingress.hostname,
                    }
                )
            elif ingress.ip != "":
                endpoints.append(
                    {
                        "name": service.name,
                        "endpoint": ingress.ip,
                    }
                )
        return endpoints
    else:
        return get_nodeport_service_endpoints(service)


def get_nodeport_service_endpoints(service):
    endpoints = []
    node_ip = get_node_ip()
    for port in service.spec.ports:
        endpoints.append(
            {
                "name": f"{service.metadata.name}/{port.name}",
                "endpoint": f"{node_ip}:{port.nodePort}",
            }
        )
    return endpoints


def get_ingress_endpoints(ingress):
    ingress_endpoints = []
    host = ""
    if ingress.status.load_balancer:
        for ing in ingress.status.load_balancer.ingress:
            if ing.hostname != "":
                host = ing.hostname
            else:
                host = ing.ip

    tlsHosts = []
    if ingress.spec.tls:
        for tls in ingress.spec.tls:
            tlsHosts.extend(tls.hosts)

    if ingress.spec.rules:
        for rule in ingress.spec.rules:
            scheme = "http"
            if rule.host in tlsHosts:
                scheme = "https"
            if rule.host != "":
                host = rule.host
            if host == "":
                continue
            if rule.http:
                for path in rule.http.paths:
                    if path.path != "":
                        ingress_endpoints.append(
                            {
                                "name": f"{ingress.name}/{path.path}",
                                "endpoint": f"{scheme}://{host}{path.path}",
                            }
                        )
                    else:
                        ingress_endpoints.append(
                            {
                                "name": f"{ingress.name}",
                                "endpoint": f"{scheme}://{host}",
                            }
                        )

    return ingress_endpoints


def get_node_ip() -> str:
    config.load_kube_config()
    core_v1 = client.CoreV1Api()
    nodes = core_v1.list_node()
    if nodes.items:
        node = nodes.items[0]
        for address in node.status.addresses:
            if address.type == "ExternalIP":
                return address.address
            elif address.type == "InternalIP":
                ip = address.address
        return ip

    raise Exception("No node found.")

k8s/tools/common/endpoint.py
seal-io/appilot
"""Tool for managing Helm charts & releases."""

k8s/tools/helm/__init__.py
seal-io/appilot
CONSTRUCT_HELM_OVERRIDED_VALUES = """
You will be provided the default values of a helm chart, and a user query describing a deployment task.

Output overrided values(in yaml) for the helm installation to satisfy the user query.
You don't need to include values that are the same as the default values.
If no values are overrided, output an empty yaml.


USER QUERY:
{query}

DEFAULT VALUES:
{default_values}

OVERRIDED VALUES:
"""

CONSTRUCT_HELM_UPGRADE_VALUES = """
You will be provided the default values of a helm chart, previous values of a helm release, and a user query describing an upgrade task.

Output values(in yaml) used for the helm upgrade to satisfy the user query.
Keep the previous values in the output unlesss it need to be changed according to the user query.
If you are not sure about some values, use the default.

USER QUERY:
{query}

DEFAULT VALUES:
{default_values}

PREVIOUS VALUES:
{previous_values}

VALUES FOR UPGRADE:
"""

k8s/tools/helm/prompt.py
seal-io/appilot
import json
import logging
import os
import subprocess
from langchain import LLMChain, PromptTemplate
from langchain.agents.tools import BaseTool
import requests
import yaml
from config import config
from langchain.schema.language_model import BaseLanguageModel
from k8s.tools.common.endpoint import (
    get_ingress_endpoints,
    get_service_endpoints,
)
from k8s.tools.helm.prompt import (
    CONSTRUCT_HELM_OVERRIDED_VALUES,
    CONSTRUCT_HELM_UPGRADE_VALUES,
)
from tools.base.tools import RequireApprovalTool
from kubernetes import config, dynamic
from kubernetes.client import api_client

logger = logging.getLogger(__name__)


def trim_default_values(input_string):
    """Trim default values of a helm chart to mitigate LLM rate limit. As a tradeoff some information is lost."""
    lines = input_string.splitlines()
    first_300_lines = "\n".join(lines[:300])

    return first_300_lines


def get_chart_default_values(chart_url: str):
    """Get default values(in yaml) of a helm chart."""
    helm_show_values_command = f"helm show values {chart_url}"

    try:
        output = subprocess.check_output(
            helm_show_values_command, shell=True, universal_newlines=True
        )

        return trim_default_values(output)
    except subprocess.CalledProcessError as e:
        return f"Helm show values failed: {e}"
    except Exception as e:
        return f"Error: {e}"


def get_helm_release_values(namespace: str, name: str):
    """Get values of a helm release."""
    helm_get_values_command = f"helm get values {name} -o yaml"

    if namespace:
        helm_get_values_command += f" --namespace {namespace}"

    try:
        output = subprocess.check_output(
            helm_get_values_command, shell=True, universal_newlines=True
        )

        return output
    except subprocess.CalledProcessError as e:
        return f"Helm get values failed: {e}"
    except Exception as e:
        return f"Error: {e}"


def searchChart(keyword: str):
    """Search helm charts in Artifact Hub. Returns a matching chart object."""
    params = {
        "facets": "false",
        "verified_publisher": "true",
        "kind": 0,
        "sort": "relevance",
        # "org": ["bitnami"],
        "ts_query_web": keyword,
    }
    response = requests.get(
        "https://artifacthub.io/api/v1/packages/search", params=params
    )
    if response.status_code >= 400:
        raise Exception("failed to search helm charts: " + response.text)

    data = response.json()
    if len(data.get("packages")) == 0:
        raise Exception("no matching helm chart found")

    chart_name = data.get("packages")[0].get("name")
    repository_name = data.get("packages")[0].get("repository").get("name")
    version = data.get("packages")[0].get("version")

    response = requests.get(
        f"https://artifacthub.io/api/v1/packages/helm/{repository_name}/{chart_name}/{version}",
    )
    chart_raw = response.json()

    chart = {}
    chart["name"] = chart_raw.get("name")
    chart["version"] = chart_raw.get("version")
    chart["description"] = chart_raw.get("description")
    chart["content_url"] = chart_raw.get("content_url")
    return chart


class SearchChartTool(BaseTool):
    """Tool to search helm charts in Artifact Hub. Returns a matching chart object."""

    name = "search_helm_chart"
    description = (
        "Search helm charts in Artifact Hub. "
        'Input should be a json string with two keys, "user_query", "keyword".'
        '"user_query" is the description of the deployment task.'
        '"keyword" is the keyword to search helm charts.'
        "Output a matching chart and overrided values for the helm deployment."
    )
    llm: BaseLanguageModel

    def _run(self, text: str) -> str:
        input = json.loads(text)
        query = input.get("user_query")
        keyword = input.get("keyword")
        chart = searchChart(keyword)

        default_values = get_chart_default_values(chart.get("content_url"))

        prompt = PromptTemplate(
            template=CONSTRUCT_HELM_OVERRIDED_VALUES,
            input_variables=["query"],
            partial_variables={"default_values": default_values},
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        overrided_values = chain.run(query).strip()
        chart["overrided_values"] = yaml.safe_load(overrided_values)
        return json.dumps(chart)


class DeployApplicationTool(RequireApprovalTool):
    """Tool to deploy an application using helm charts."""

    name = "deploy_application"
    description = (
        "Deploy an application using helm charts."
        'Input should be a json string with four keys, "namespace", "name", "chart_url", "values".'
        '"namespace" is the namespace to deploy the application.'
        '"name" is the name of the application, generate a reasonable one if not specified.'
        '"chart_url" is the url to download the helm chart.'
        '"values" is overrided values for the helm installation to satisfy user query.'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        chart_url = input.get("chart_url")
        name = input.get("name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"

        helm_install_command = (
            f"helm install {name} {chart_url}  --namespace {namespace}"
        )

        values = input.get("values")
        if values:
            values = {}

        # add chart_url to values as metadata until https://github.com/helm/helm/issues/4256 is resolved.
        if "global" in values:
            values["global"]["metadata_chart_url"] = chart_url
        else:
            values["global"] = {"metadata_chart_url": chart_url}

        output_directory = "/tmp/appilot"
        if not os.path.exists(output_directory):
            os.makedirs(output_directory)
        file_path = f"{output_directory}/values.yaml"
        with open(file_path, "w") as file:
            yaml.dump(input.get("values"), file)
        helm_install_command += f" -f {file_path}"

        try:
            output = subprocess.check_output(
                helm_install_command, shell=True, universal_newlines=True
            )

            logger.debug(f"helm install output: {output}")
            return f"application {name} is deployed."
        except subprocess.CalledProcessError as e:
            return f"Helm install failed: {e}"
        except Exception as e:
            return f"Error: {e}"


class GenerateUpgradeApplicationValuesTool(BaseTool):
    """Tool to generate values for upgrading an application."""

    name = "generate_upgrade_application_values"
    description = (
        "Generate values for upgrading an application deployed by helm chart."
        'Input should be a json string with three keys, "namespace", "name", "user_query".'
        '"namespace" is the namespace of the application.'
        '"name" is the name of the application.'
        '"user_query" is the description of the deployment task.'
        "Output overrided values for the helm upgrade."
    )
    llm: BaseLanguageModel

    def _run(self, text: str) -> str:
        input = json.loads(text)
        namespace = input.get("namespace")
        name = input.get("name")
        query = input.get("user_query")

        if namespace == "":
            namespace = "default"

        previous_values = get_helm_release_values(namespace, name)

        chart_url = (
            yaml.safe_load(previous_values)
            .get("global", {})
            .get("metadata_chart_url")
        )
        if not chart_url:
            return "Missing chart_url metadata in previous release"

        default_values = get_chart_default_values(chart_url)

        prompt = PromptTemplate(
            template=CONSTRUCT_HELM_UPGRADE_VALUES,
            input_variables=["query"],
            partial_variables={
                "default_values": default_values,
                "previous_values": previous_values,
            },
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        overrided_values_yaml = chain.run(query).strip()

        overrided_values = yaml.safe_load(overrided_values_yaml)
        if (
            "global" in overrided_values
            and "metadata_chart_url" in overrided_values["global"]
        ):
            del overrided_values["global"]["metadata_chart_url"]

        return json.dumps(overrided_values)


class UpgradeApplicationTool(RequireApprovalTool):
    """Tool to upgrade an application."""

    name = "upgrade_application"
    description = (
        "Upgrade an application."
        'Input should be a json string with three keys, "namespace", "name", "values".'
        '"namespace" is the namespace to deploy the application.'
        '"name" is the name of the application, generate a reasonable one if not specified.'
        '"values" is overrided values for helm upgrade to satisfy user query.'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        namespace = input.get("namespace")
        name = input.get("name")
        values = input.get("values")

        previous_values = get_helm_release_values(namespace, name)

        chart_url = (
            yaml.safe_load(previous_values)
            .get("global", {})
            .get("metadata_chart_url")
        )
        if not chart_url:
            return "Missing chart_url metadata in previous release"

        if namespace == "":
            namespace = "default"

        helm_upgrade_command = (
            f"helm upgrade {name} {chart_url} --namespace {namespace}"
        )

        if values:
            # add chart_url to values as metadata until https://github.com/helm/helm/issues/4256 is resolved.
            if "global" in values:
                values["global"]["metadata_chart_url"] = chart_url
            else:
                values["global"] = {"metadata_chart_url": chart_url}

            output_directory = "/tmp/appilot"
            if not os.path.exists(output_directory):
                os.makedirs(output_directory)
            file_path = f"{output_directory}/values.yaml"
            with open(file_path, "w") as file:
                yaml.dump(values, file)
            helm_upgrade_command += f" -f {file_path}"

        try:
            output = subprocess.check_output(
                helm_upgrade_command, shell=True, universal_newlines=True
            )

            logger.debug(f"helm upgrade output: {output}")
            return f"application {name} is upgraded."
        except subprocess.CalledProcessError as e:
            return f"Helm upgrade failed: {e}"
        except Exception as e:
            return f"Error: {e}"


def get_pod_ready_status_of_helm_release(name: str, namespace: str) -> str:
    if namespace == "":
        namespace = "default"

    helm_list_command = f"helm get manifest {name} --namespace {namespace}"

    try:
        output = subprocess.check_output(
            helm_list_command, shell=True, universal_newlines=True
        )
    except subprocess.CalledProcessError as e:
        return f"Failed to get helm manifest: {e}"
    except Exception as e:
        return f"Error: {e}"

    resource_manifests = yaml.safe_load_all(output)

    dyn_client = dynamic.DynamicClient(
        api_client.ApiClient(configuration=config.load_kube_config())
    )

    replicas = 0
    ready_replicas = 0
    for resource_manifest in resource_manifests:
        if not resource_manifest:
            continue
        resource_kind = resource_manifest.get("kind")
        if resource_kind not in ["Deployment", "StatefulSet", "DaemonSet"]:
            continue

        resoruce_client = dyn_client.resources.get(
            api_version=resource_manifest.get("apiVersion"),
            kind=resource_kind,
        )
        resource = resoruce_client.get(
            name=resource_manifest.get("metadata").get("name"),
            namespace=namespace,
        )

        if resource_kind == "Deployment":
            replicas += resource.get("spec").get("replicas", 0)
            ready_replicas += resource.get("status").get("readyReplicas", 0)
        elif resource_kind == "StatefulSet":
            replicas += resource.get("spec").get("replicas", 0)
            ready_replicas += resource.get("status").get("readyReplicas", 0)
        elif resource_kind == "DaemonSet":
            replicas += resource.get("status").get("desiredNumberScheduled", 0)
            ready_replicas += resource.get("status").get("numberReady", 0)

    return f"{ready_replicas}/{replicas}"


def get_resource_pods(dyn_client, namespace, resource):
    selector = resource.spec.selector.match_labels
    pod_client = dyn_client.resources.get(
        api_version="v1",
        kind="Pod",
    )
    pods = pod_client.get(
        namespace=namespace,
        label_selector=selector,
    )
    return pods.to_dict()


def tidy_up_resource(resource):
    try:
        # make prompt short.
        del resource["metadata"]["managedFields"]
        del resource["metadata"]["resourceVersion"]
        del resource["metadata"]["uid"]
        del resource["metadata"]["generation"]
    except KeyError:
        pass


class ListApplicationsTool(BaseTool):
    """Tool to list applications."""

    name = "list_applications"
    description = (
        "List applications."
        'Input should be a json string with one keys, "namespace".'
        "namespace can be empty if not specified. If namespace is empty, list in current namespace."
        "If namespace is --all, list applications in all namespaces."
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        namespace = input.get("namespace")

        helm_list_command = "helm list --all -o json"

        if namespace == "--all":
            helm_list_command += " --all-namespaces"
        else:
            if namespace == "":
                namespace = "default"
            helm_list_command += f" --namespace {namespace}"

        try:
            output = subprocess.check_output(
                helm_list_command, shell=True, universal_newlines=True
            )

        except subprocess.CalledProcessError as e:
            return f"Helm list failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        helm_releases = json.loads(output)
        for helm_release in helm_releases:
            del helm_release["chart"]
            del helm_release["app_version"]
            helm_release["ready"] = get_pod_ready_status_of_helm_release(
                helm_release.get("name"), helm_release.get("namespace")
            )
        return json.dumps(helm_releases)


class GetApplicationResourcesTool(BaseTool):
    """Tool to get application resources."""

    name = "get_application_resources"
    description = (
        "Get application resources. "
        "Helpful to know what resources an application consists of, what status they are in."
        'Input should be a json string with two keys: "name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        name = input.get("name")
        namespace = input.get("namespace")

        if namespace == "":
            namespace = "default"

        helm_manifest_command = (
            f"helm get manifest {name} --namespace {namespace}"
        )

        try:
            output = subprocess.check_output(
                helm_manifest_command, shell=True, universal_newlines=True
            )

        except subprocess.CalledProcessError as e:
            return f"Helm delete failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        resource_manifests = yaml.safe_load_all(output)

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )

        resources = []
        for resource_manifest in resource_manifests:
            resource_kind = resource_manifest.get("kind")
            resoruce_client = dyn_client.resources.get(
                api_version=resource_manifest.get("apiVersion"),
                kind=resource_kind,
            )
            resource = resoruce_client.get(
                name=resource_manifest.get("metadata").get("name"),
                namespace=namespace,
            )
            if resource_kind in ["Deployment", "StatefulSet", "DaemonSet"]:
                pods = get_resource_pods(dyn_client, namespace, resource)
                resources.extend(pods)
            resource = resource.to_dict()
            tidy_up_resource(resource)
            resources.append(resource)

        return json.dumps(resources)


class GetApplicationAccessEndpointsTool(BaseTool):
    """Tool to get application access endpoints."""

    name = "get_application_access_endpoints"
    description = (
        "Get application access endpoints. "
        'Input should be a json string with two keys: "name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        name = input.get("name")
        namespace = input.get("namespace")

        helm_manifest_command = f"helm get manifest {name}"

        if namespace == "":
            namespace = "default"

        helm_manifest_command += f" --namespace {namespace}"

        try:
            output = subprocess.check_output(
                helm_manifest_command, shell=True, universal_newlines=True
            )

        except subprocess.CalledProcessError as e:
            return f"Helm delete failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        resource_manifests = yaml.safe_load_all(output)

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )

        endpoints = []
        for resource_manifest in resource_manifests:
            resource_kind = resource_manifest.get("kind")
            if resource_kind not in ["Service", "Ingress"]:
                continue

            resoruce_client = dyn_client.resources.get(
                api_version=resource_manifest.get("apiVersion"),
                kind=resource_kind,
            )
            resource = resoruce_client.get(
                name=resource_manifest.get("metadata").get("name"),
                namespace=namespace,
            )

            if resource_kind == "Service":
                endpoints.extend(get_service_endpoints(resource))
            elif resource_kind == "Ingress":
                endpoints.extend(get_ingress_endpoints(resource))

        return json.dumps(endpoints)


class GetApplicationDetailTool(BaseTool):
    """Tool to get application detail."""

    name = "get_application_detail"
    description = (
        "Get application detail. "
        'Input should be a json string with two keys: "name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        name = input.get("name")
        namespace = input.get("namespace")

        if namespace == "":
            namespace = "default"

        helm_status_command = (
            f"helm status {name} --show-resources --namespace {namespace}"
        )

        try:
            output = subprocess.check_output(
                helm_status_command, shell=True, universal_newlines=True
            )

        except subprocess.CalledProcessError as e:
            return f"Helm status failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        # Trim NOTES
        index = output.find("NOTES:")

        if index != -1:
            output = output[:index]

        return output


class DeleteApplicationTool(RequireApprovalTool):
    """Tool to delete an application."""

    name = "delete_application"
    description = (
        "Delete an application. "
        'Input should be a json string with two keys: "name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)
        name = input.get("name")
        namespace = input.get("namespace")

        if namespace == "":
            namespace = "default"

        helm_delete_command = f"helm delete {name} --namespace {namespace}"

        try:
            output = subprocess.check_output(
                helm_delete_command, shell=True, universal_newlines=True
            )

            logger.debug(f"helm delete output: {output}")
            return "Application is deleted."
        except subprocess.CalledProcessError as e:
            return f"Helm delete failed: {e}"
        except Exception as e:
            return f"Error: {e}"

k8s/tools/helm/tool.py
seal-io/appilot
"""Tool for managing Kubernetes resources."""

k8s/tools/manage_resource/__init__.py
seal-io/appilot
CONSTRUCT_RESOURCES_TO_CREATE_PROMPT = """
You are a planner that constructs kubernetes resources given a user query describing a deployment task.

You should:
1) evaluate whether kubernetes resources can be constructed according to the user query. If no, say why.
2) if yes, output in the following format:

CONSTRUCTED RESOURCES: 
<KUBERNETES_RESOURCES_IN_YAML>

Strictly follow the above output format, do not add extra explanation or words.
The output will be applied to a kubernetes cluster for creation.
Ensure the output has no placeholders and requires no external inputs.

User query: {query}

CONSTRUCTED RESOURCES: 
"""

CONSTRUCT_RESOURCES_TO_UPDATE_PROMPT = """
You are a planner that constructs kubernetes resources given a user query describing an update task.

You should:
1) evaluate whether kubernetes resources can be constructed according to the user query. If no, say why.
2) if yes, output in the following format:

CURRENT RESOURCE SPEC:
<CURRENT_RESOURCE_SPEC_IN_YAML>

TO UPDATE RESOURCE SPEC:
<TO_UPDATE_RESOURCE_SPEC_IN_YAML>

Strictly follow the above output format, do not add extra explanation or words.
The output will be applied to a kubernetes cluster for creation.

User query: {query}

CURRENT RESOURCE SPEC:
{current_resource_spec}

TO UPDATE RESOURCE SPEC:
"""

k8s/tools/manage_resource/prompt.py
seal-io/appilot
import json
import logging
import subprocess
import click
from langchain import LLMChain, PromptTemplate
from langchain.agents.tools import BaseTool
import yaml
from config import config
from langchain.schema.language_model import BaseLanguageModel
from k8s.tools.common.endpoint import (
    get_ingress_endpoints,
    get_service_endpoints,
)
from k8s.tools.manage_resource.prompt import (
    CONSTRUCT_RESOURCES_TO_CREATE_PROMPT,
    CONSTRUCT_RESOURCES_TO_UPDATE_PROMPT,
)
from tools.base.tools import RequireApprovalTool
from kubernetes import config, dynamic, client
from kubernetes.client import api_client
from k8s import context
from utils import utils
from i18n import text

logger = logging.getLogger(__name__)


class ListResourcesTool(BaseTool):
    """Tool to list resources."""

    name = "list_kubernetes_resources"
    description = (
        "List kubernetes resources."
        'Input should be a json string with two keys: "resource_kind" and "namespace".'
        '"namespace" is optional, set it to empty string if user does not specify.'
        "If namespace is --all, lists in all namespaces."
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        resource_kind = str(input.get("resource_kind")).lower()
        namespace = str(input.get("namespace")).lower()

        # Use kubectl directly. Raw API output can easily exceed the LLM rate limit.
        kubectl_get_command = f"kubectl get {resource_kind}"

        if namespace == "--all":
            kubectl_get_command += " --all-namespaces"
        elif namespace:
            kubectl_get_command += f" -n {namespace}"

        try:
            output = subprocess.check_output(
                kubectl_get_command, shell=True, universal_newlines=True
            )

        except subprocess.CalledProcessError as e:
            return f"kubectl get failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        # Print raw output without markdown rendering
        return f"{utils.raw_format_prefix}\n{output}"


class ListResourcesForInfoTool(ListResourcesTool):
    """Tool to list resources for info."""

    name = "list_kubernetes_resources_for_info"
    description = (
        "List kubernetes resources to help find more info, instead of returning to the user."
        'Input should be a json string with two keys: "resource_kind" and "namespace".'
        '"namespace" is optional, set it to empty string if user does not specify.'
        "If namespace is --all, lists in all namespaces."
    )


class DeleteResourceTool(RequireApprovalTool):
    """Tool to delete a kubernetes resource."""

    name = "delete_a_kubernetes_resource"
    description = (
        "Delete a kubernetes resource. "
        'Input should be a json string with three keys: "resource_kind", "resource_name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )
        resource_kind = input.get("resource_kind")
        resource_name = input.get("resource_name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"
        gvk = context.search_api_resource(resource_kind)
        resources = dyn_client.resources.get(
            api_version=gvk.groupVersion,
            kind=gvk.kind,
        )

        try:
            resources.delete(name=resource_name, namespace=namespace)
        except Exception as e:
            return f"Error deleting resource: {e}"

        return "Resource is being deleted."


class GetResourceDetailTool(BaseTool):
    """Tool to get detail of a kubernetes resource."""

    name = "get_kubernetes_resource_detail"
    description = (
        "Get detail of a kubernetes resource. "
        'Input should be a json string with three keys: "resource_kind", "resource_name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )
        resource_kind = input.get("resource_kind")
        resource_name = input.get("resource_name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"
        gvk = context.search_api_resource(resource_kind)
        resources = dyn_client.resources.get(
            api_version=gvk.groupVersion,
            kind=gvk.kind,
        )

        try:
            resource_raw = resources.get(
                name=resource_name, namespace=namespace
            )
            resource = resource_raw.to_dict()
            # make prompt short.
            del resource["metadata"]["managedFields"]
            del resource["metadata"]["resourceVersion"]
            del resource["metadata"]["uid"]
            del resource["metadata"]["generation"]
        except KeyError:
            pass
        except Exception as e:
            return f"Error getting resource detail: {e}"

        return json.dumps(resource)


class GetResourceYamlTool(BaseTool):
    """Tool to get yaml of a kubernetes resource."""

    name = "get_kubernetes_resource_yaml"
    description = (
        "Get yaml of a kubernetes resource. "
        'Input should be a json string with three keys: "resource_kind", "resource_name" and "namespace".'
        "The output is the yaml of the resource."
        "It directly prints the output. Use when users want to see the resoruce yaml."
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )
        resource_kind = input.get("resource_kind")
        resource_name = input.get("resource_name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"
        gvk = context.search_api_resource(resource_kind)
        resources = dyn_client.resources.get(
            api_version=gvk.groupVersion,
            kind=gvk.kind,
        )

        try:
            resource_raw = resources.get(
                name=resource_name, namespace=namespace
            )
            resource = resource_raw.to_dict()
            # make prompt short.
            del resource["metadata"]["managedFields"]
            del resource["metadata"]["resourceVersion"]
            del resource["metadata"]["uid"]
            del resource["metadata"]["generation"]
        except KeyError:
            pass
        except Exception as e:
            logger.debug(f"Error getting resource yaml: {e}")
            # FIXME handle other cases
            return f"Resource not found."

        # Print raw output without markdown rendering
        return f"{utils.raw_format_prefix}\n{yaml.dump(resource)}"


class GetServiceAccessEndpointsTool(BaseTool):
    """Tool to get access endpoints of a kubernetes service."""

    name = "get_kubernetes_service_access_endpoints"
    description = (
        "Get access endpoints of a kubernetes service. "
        'Input should be a json string with two keys: "namespace" and "name".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        name = input.get("name")
        namespace = input.get("namespace")

        if namespace == "":
            namespace = "default"

        try:
            dynamic_client = dynamic.DynamicClient(
                client.api_client.ApiClient()
            )
            resource = dynamic_client.resources.get(
                api_version="v1", kind="Service"
            )
            service = resource.get(namespace=namespace, name=name)

            endpoints = get_service_endpoints(service)
        except Exception as e:
            return f"Error getting service endpoints: {e}"

        return json.dumps(endpoints)


class GetIngressAccessEndpointsTool(BaseTool):
    """Tool to get access endpoints of a kubernetes ingress."""

    name = "get_kubernetes_ingress_access_endpoints"
    description = (
        "Get access endpoints of a kubernetes ingress. "
        'Input should be a json string with two keys: "namespace" and "name".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        name = input.get("name")
        namespace = input.get("namespace")

        if namespace == "":
            namespace = "default"

        try:
            dynamic_client = dynamic.DynamicClient(
                client.api_client.ApiClient()
            )
            resource = dynamic_client.resources.get(
                api_version="networking.k8s.io/v1", kind="Ingress"
            )
            ingress = resource.get(namespace=namespace, name=name)

            endpoints = get_ingress_endpoints(ingress)
        except Exception as e:
            return f"Error getting ingress endpoints: {e}"

        return json.dumps(endpoints)


class DescribePodTool(BaseTool):
    """Tool to describe a pod."""

    name = "describe_pod"
    description = (
        "Show details of a pod including related events. "
        'Input should be a json string with two keys: "name" and "namespace".'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        name = input.get("name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"

        kubectl_describe_command = (
            f"kubectl describe pod {name} -n {namespace}"
        )

        try:
            output = subprocess.check_output(
                kubectl_describe_command, shell=True, universal_newlines=True
            )
        except subprocess.CalledProcessError as e:
            return f"kubectl get failed: {e}"
        except Exception as e:
            return f"Error: {e}"

        return output


class GetPodLogsTool(BaseTool):
    """Tool to get logs of a pod."""

    name = "get_kubernetes_pod_logs"
    description = (
        "Get logs of a pod. "
        'Input should be a json string with four keys: "name", "namespace", "container_name" and "line_number".'
        '"container_name" is optional. Set it to the name of the container to get logs from.'
        '"line_number" is int defaulting to 50. Set it to the number of lines of logs to return.'
    )

    def _run(self, text: str) -> str:
        input = json.loads(text)

        name = input.get("name")
        namespace = input.get("namespace")
        container_name = input.get("container_name", "")
        line_number = input.get("line_number", 50)

        if namespace == "":
            namespace = "default"

        v1 = client.CoreV1Api()

        try:
            pod_log = v1.read_namespaced_pod_log(
                name=name,
                namespace=namespace,
                container=container_name,
                tail_lines=line_number,
            )
        except Exception as e:
            return f"Error getting pod logs: {e}"

        return f"```\n{pod_log}\n```"


class WatchResourcesTool(BaseTool):
    """Tool to watch resources."""

    name = "watch_resources"
    description = (
        "Watch resources changes in a namespace."
        'Input should be a json string with two keys: "resource_kind" and "namespace".'
    )

    def _run(self, query: str) -> str:
        input = json.loads(query)
        resource_kind = str(input.get("resource_kind")).lower()
        namespace = str(input.get("namespace")).lower()

        if namespace == "":
            namespace = "default"

        kubectl_watch_command = (
            f"kubectl get {resource_kind} -w -n {namespace}"
        )

        try:
            click.echo(text.get("watch_service_note"))
            process = subprocess.Popen(
                kubectl_watch_command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )

            for line in process.stdout:
                print(line, end="")

            process.wait()

        except KeyboardInterrupt:
            # Ctrl+C detected. Stopping the request.
            print("Terminated by user")

        return text.get("watch_service_ending")


class ConstructResourceTool(BaseTool):
    """Tool to construct resources."""

    name = "construct_kubernetes_resources"
    description = (
        "Construct Kubernetes resources. "
        'Input to the tool should be a json with one keys: "user_query".'
        'The value of "user_query" should be the description of a deployment task.'
        "The output is kubernetes resources in yaml format."
    )
    llm: BaseLanguageModel

    def _run(self, text: str) -> str:
        input = json.loads(text)
        query = input.get("user_query")
        prompt = PromptTemplate(
            template=CONSTRUCT_RESOURCES_TO_CREATE_PROMPT,
            input_variables=["query"],
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        return chain.run(json.dumps(query)).strip()


class ConstructResourceForUpdateTool(BaseTool):
    """Tool to construct a kubernetes resource for update."""

    name = "construct_kubernetes_resource_for_update"
    description = (
        "Construct a Kubernetes resource for update. "
        'Input to the tool should be a json with four keys: "user_query", "resource_kind", "resource_name" and "namespace".'
        '"user_query" should be the description of an update task.'
        "The output is the kubernetes resource spec for update in yaml format."
    )
    llm: BaseLanguageModel

    def _run(self, text: str) -> str:
        input = json.loads(text)
        query = input.get("user_query")
        resource_kind = input.get("resource_kind")
        resource_name = input.get("resource_name")
        namespace = input.get("namespace")
        if namespace == "":
            namespace = "default"

        dyn_client = dynamic.DynamicClient(
            api_client.ApiClient(configuration=config.load_kube_config())
        )
        gvk = context.search_api_resource(resource_kind)
        resources = dyn_client.resources.get(
            api_version=gvk.groupVersion,
            kind=gvk.kind,
        )

        try:
            resource_raw = resources.get(
                name=resource_name, namespace=namespace
            )
            resource = resource_raw.to_dict()
            # make prompt short.
            del resource["metadata"]["managedFields"]
            del resource["metadata"]["resourceVersion"]
            del resource["metadata"]["uid"]
            del resource["metadata"]["generation"]
            del resource["status"]
        except KeyError:
            pass
        except Exception as e:
            return f"Error getting resource detail: {e}"

        prompt = PromptTemplate(
            template=CONSTRUCT_RESOURCES_TO_UPDATE_PROMPT,
            input_variables=["query"],
            partial_variables={"current_resource_spec": yaml.dump(resource)},
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        return chain.run(json.dumps(query)).strip()


class ApplyResourcesTool(RequireApprovalTool):
    """Tool to apply kubernetes resources."""

    name = "apply_kubernetes_resources"
    description = (
        "Apply kubernetes resources. "
        "Input should be kubernetes resources in yaml format."
        "The provided yaml should is fully functional. Do not provide empty input. "
        "It will be applied to a kubernetes cluster."
    )

    def _run(self, text: str) -> str:
        lines = text.splitlines()
        # remove triple backticks of the yaml text
        filtered_lines = [line for line in lines if not line.startswith("```")]
        filtered_text = "\n".join(filtered_lines)
        try:
            yaml_documents = yaml.safe_load_all(filtered_text)
            apply_or_update_yaml(yaml_documents)
        except Exception as e:
            return f"Error applying/updating YAML manifest: {str(e)}"

        resources = []
        for yaml_manifest in yaml_documents:
            resources.append(
                {
                    "kind": yaml_manifest["kind"],
                    "name": yaml_manifest["metadata"]["name"],
                }
            )

        return f"Applied the following resources: {resources}"


def apply_or_update_yaml(yaml_documents):
    for yaml_manifest in yaml_documents:
        api_version = yaml_manifest["apiVersion"]
        kind = yaml_manifest["kind"]

        config.load_kube_config()
        dynamic_client = dynamic.DynamicClient(client.api_client.ApiClient())
        resource = dynamic_client.resources.get(
            api_version=api_version, kind=kind
        )

        namespace = yaml_manifest.get("metadata", {}).get(
            "namespace", "default"
        )
        name = yaml_manifest.get("metadata", {}).get("name", "")

        resource_exists = False
        try:
            resource.get(namespace=namespace, name=name)
            resource_exists = True
        except Exception as e:
            pass

        # update if exists, else create.
        try:
            if resource_exists:
                resource.patch(
                    namespace=namespace,
                    name=name,
                    body=yaml_manifest,
                )
            else:
                resource.create(namespace=namespace, body=yaml_manifest)
        except Exception as e:
            print(f"Error applying/updating YAML manifest: {str(e)}")

k8s/tools/manage_resource/tool.py
seal-io/appilot
from typing import Any
from langchain.agents.tools import BaseTool
from callbacks.handlers import ApprovalCallbackHandler


class RequireApprovalTool(BaseTool):
    """Tool that requires human approval."""

    def __init__(self, **data: Any) -> None:
        super().__init__(callbacks=[ApprovalCallbackHandler()], **data)

tools/base/tools.py
seal-io/appilot
"""Tool for asking human input."""

tools/human/__init__.py
seal-io/appilot
from typing import Callable, Optional

from pydantic import Field

from langchain.callbacks.manager import CallbackManagerForToolRun
from langchain.tools.base import BaseTool


def _print_func(text: str) -> None:
    print("\n")
    print(text)


class HumanTool(BaseTool):
    """Tool that asks user for input."""

    name = "human"
    description = (
        "You can ask a human for guidance when you think you "
        "got stuck or you are not sure what to do next. "
        "The input should be a question for the human."
    )
    prompt_func: Callable[[str], None] = Field(
        default_factory=lambda: _print_func
    )
    input_func: Callable = Field(default_factory=lambda: input)

    def _run(
        self,
        query: str,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """Use the Human input tool."""
        self.prompt_func(query)
        return self.input_func(">")

tools/human/tool.py
seal-io/appilot
"""Tool for switching Appilot reasoning output."""

tools/reasoning/__init__.py
seal-io/appilot
from config import config

from langchain.tools.base import BaseTool


class ShowReasoningTool(BaseTool):
    """Tool that show Appilot reasoning output."""

    name = "show_reasoning_output"
    description = "Show Appilot reasoning output."

    def _run(self, query: str) -> str:
        config.set_show_reasoning(True)
        return "succeed."


class HideReasoningTool(BaseTool):
    """Tool that hide Appilot reasoning output."""

    name = "hide_reasoning_output"
    description = "Hide Appilot reasoning output."

    def _run(self, query: str) -> str:
        config.set_show_reasoning(False)
        return "succeed."

tools/reasoning/tool.py
seal-io/appilot
import os
import sys

from i18n import text

from colorama import Fore, Style
from rich.markdown import Markdown
from rich.console import Console
from datetime import datetime
from dateutil import parser
from datetime import timezone

console = Console()

inform_sent = False

raw_format_prefix = "[raw]"


def get_env(key: str, default: str = "") -> str:
    env = os.getenv(key)
    if env is None:
        return default
    return env.strip()


def get_env_list(key: str, default: [] = []) -> list[str]:
    env = os.getenv(key)
    if env is None:
        return default
    return [item.strip() for item in env.split(",")]


def get_env_bool(key: str, default: bool = False) -> bool:
    env = os.getenv(key)
    if env is None:
        return default
    else:
        return env.lower() in ["1", "true", "yes", "on"]


def print_ai_reasoning(message):
    print(Fore.CYAN + text.get("ai_reasoning") + message + Style.RESET_ALL)


def print_ai_inform(message):
    global inform_sent
    inform_sent = True
    # move cursor to the end of previous line
    sys.stdout.write("\033[F\033[1000C")
    print(
        Fore.LIGHTYELLOW_EX
        + "\n"
        + text.get("inform_prefix")
        + message
        + Style.RESET_ALL
    )


def is_inform_sent():
    global inform_sent
    if inform_sent:
        inform_sent = False
        return True


def print_ai_response(message):
    print(text.get("response_prefix"), end="")
    if message.startswith(raw_format_prefix):
        print(message[len(raw_format_prefix) :])
    else:
        console.print(Markdown(message))


def print_rejected_message():
    print(text.get("response_prefix"), end="")
    print(text.get("rejected_message"))


def format_relative_time(iso_time):
    parsed_time = parser.isoparse(iso_time)
    current_time = datetime.now(timezone.utc)
    time_difference = current_time - parsed_time

    days = time_difference.days
    hours = int(time_difference.total_seconds() // 3600)
    minutes = int((time_difference.total_seconds() % 3600) // 60)

    if days > 0:
        return f"{days} Days ago"
    elif hours > 0:
        return f"{hours} Hours ago"
    elif minutes > 0:
        return f"{minutes} Minutes ago"
    else:
        return "Just now"

utils/utils.py
seal-io/appilot
import json
from typing import List
import click
import requests
from i18n import text
from utils import utils


class WalrusClient:
    """HTTP client for Walrus API."""

    def __init__(self, api_url: str, api_key: str, **kwargs):
        self.api_url = api_url
        self.api_key = api_key
        self.request_args = kwargs

    def headers(self):
        """Get default headers."""
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

    def list_projects(self):
        """List projects."""
        response = requests.get(
            url=self.api_url + "/v1/projects",
            params={"perPage": -1},
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to list projects: {response.text}")

        return response.json()["items"]

    def get_project(self, project: str):
        """Get a project by id or name."""
        response = requests.get(
            url=self.api_url + f"/v1/projects/{project}",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get project {project}: {response.text}"
            )

        return response.json()

    def list_environments(self, project_id: str):
        """List environments."""
        params = {
            "perPage": -1,
        }
        response = requests.get(
            url=self.api_url + f"/v1/projects/{project_id}/environments",
            params=params,
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to list environments: {response.text}")

        return response.json()["items"]

    def get_environment(self, project_id: str, environment: str):
        """Get an environment by id or name."""
        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment}",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get environment {environment}: {response.text}"
            )

        return response.json()

    def create_environment(self, project_id: str, data):
        """Create an environment in a project."""

        response = requests.post(
            url=self.api_url + f"/v1/projects/{project_id}/environments",
            headers=self.headers(),
            json=data,
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to create environment: {response.text}")

        return response.text

    def delete_environments(self, project_id: str, ids: List[str]):
        """Delete one or multiple environments."""
        items = []
        for id in ids:
            items.append({"id": id})

        body = {"items": items}
        response = requests.delete(
            url=self.api_url + f"/v1/projects/{project_id}/environments",
            headers=self.headers(),
            json=body,
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to delete environment: {response.text}")

        return response.text

    def get_environment_graph(self, project_id: str, environment_id: str):
        """Get environment dependency graph."""
        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/graph",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get environment dependency graph: {response.text}"
            )

        return response.json()

    def list_services(self, project_id: str, environment_id: str):
        """List services in a project and environment."""
        params = {
            "perPage": -1,
        }

        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services",
            params=params,
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to list services: {response.text}")

        return response.json()["items"]

    def watch_services(self, project_id: str, environment_id: str):
        """Watch services in a project and environment."""

        def align_and_echo(data_list, width=30):
            aligned_data = [item.ljust(width) for item in data_list]
            click.echo("".join(aligned_data))

        def print_service(s):
            align_and_echo(
                [
                    s.get("name"),
                    s.get("template").get("name"),
                    s.get("status").get("summaryStatus"),
                    utils.format_relative_time(s.get("createTime")),
                ]
            )

        services = self.list_services(project_id, environment_id)
        click.echo(text.get("watch_service_note"))
        align_and_echo(
            [
                "NAME",
                "TEMPLATE",
                "STATUS",
                "CREATE TIME",
            ]
        )
        for service in services:
            print_service(service)

        params = {
            "perPage": -1,
            "watch": "true",
        }

        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services",
            params=params,
            headers=self.headers(),
            **self.request_args,
            stream=True,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to list services: {response.text}")

        for chunk in response.iter_content(chunk_size=None):
            event = json.loads(chunk.decode("utf-8"))
            if "items" in event:
                for item in event["items"]:
                    print_service(item)

    def list_services_in_all_environments(self, project_id: str):
        """List services in all environments of a project."""
        params = {
            "perPage": -1,
        }

        services = []
        envs = self.list_environments(project_id)
        for env in envs:
            response = requests.get(
                url=self.api_url
                + f"/v1/projects/{project_id}/environments/{env['id']}/services",
                params=params,
                headers=self.headers(),
                **self.request_args,
            )
            if response.status_code >= 400:
                raise Exception(f"Failed to list services: {response.text}")

            services_in_env = response.json()["items"]
            if services_in_env is None or len(services_in_env) == 0:
                continue

            services.extend(response.json()["items"])

        return services

    def get_service_by_name(
        self, project_id: str, environment_id: str, service_name: str
    ):
        """Get a service by name."""

        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service_name}",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to get service: {response.text}")

        return response.json()

    def create_service(self, project_id: str, environment_id: str, data):
        """Create a service in a project and environment."""

        response = requests.post(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services",
            headers=self.headers(),
            json=data,
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to create service: {response.text}")

        return response.text

    def update_service(self, project_id: str, environment_id: str, data):
        """Update a service in a project and environment."""
        try:
            service = json.loads(data)
        except json.JSONDecodeError as e:
            raise e

        response = requests.put(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service['id']}/upgrade",
            headers=self.headers(),
            data=data,
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to update service: {response.text}")

        return response.text

    def delete_services(
        self, project_id: str, environment_id: str, ids: List[str]
    ):
        """Delete one or multiple services."""
        items = []
        for id in ids:
            items.append({"id": id})

        body = {"items": items}
        response = requests.delete(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services",
            headers=self.headers(),
            json=body,
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to delete service: {response.text}")

        return response.text

    def get_service_access_endpoints(
        self, project_id: str, environment_id: str, service_id: str
    ):
        """Get access endpoints of a service."""
        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service_id}/access-endpoints",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get service access endpoints: {response.text}"
            )

        return response.text

    def list_service_resources(
        self, project_id: str, environment_id: str, service_id: str
    ):
        """List resources of a service."""
        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service_id}/resources",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get service resources: {response.text}"
            )

        return response.json()["items"]

    def get_service_resource_keys(
        self,
        project_id: str,
        environment_id: str,
        service_id: str,
        service_resource_id: str,
    ):
        """Get keys of a service resource."""
        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service_id}/resources/{service_resource_id}/keys",
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get service resource keys: {response.text}"
            )

        return response.text

    def get_service_resource_logs(
        self,
        project_id: str,
        environment_id: str,
        service_id: str,
        service_resource_id: str,
        key: str,
        line_number: int,
    ):
        """Get logs of a service resource."""
        params = {
            "key": key,
            "tailLines": line_number,
        }

        response = requests.get(
            url=self.api_url
            + f"/v1/projects/{project_id}/environments/{environment_id}/services/{service_id}/resources/{service_resource_id}/log",
            params=params,
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to get service resource logs: {response.text}"
            )

        return response.text

    def list_templates(self):
        """List templates."""
        response = requests.get(
            url=self.api_url + "/v1/templates",
            params={"perPage": -1},
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(f"Failed to list templates: {response.text}")

        templates = response.json()["items"]
        for template in templates:
            del template["createTime"]
            del template["updateTime"]
            del template["status"]
            del template["source"]
        return json.dumps(templates)

    def get_template_version(self, template: str):
        """Get latest template version given template id or name."""
        response = requests.get(
            url=self.api_url + f"/v1/templates/{template}/versions",
            params={"perPage": -1},
            headers=self.headers(),
            **self.request_args,
        )
        if response.status_code >= 400:
            raise Exception(
                f"Failed to list versions of template {template}: {response.text}"
            )

        template_versions = response.json()["items"]
        if len(template_versions) == 0:
            raise Exception("Template version not found")

        keys_to_remove = [
            "readme",
            "outputs",
            "requiredProviders",
            "createTime",
            "updateTime",
            "id",
            "source",
        ]

        template_version = response.json()["items"][0]
        # remove keys that are not needed to make prompt neat

        for key in keys_to_remove:
            if key in template_version:
                del template_version[key]
            if key in template_version["schema"]:
                del template_version["schema"][key]

        return json.dumps(template_version)

walrus/client.py
seal-io/appilot
from pydantic import BaseModel

from walrus.client import WalrusClient


class Context(BaseModel):
    project_id: str = ""
    project_name: str = ""
    environment_id: str = ""
    environment_name: str = ""


GLOBAL_CONTEXT: Context


def set_default(
    walrus_url: str,
    walrus_api_key: str,
    default_project: str = "",
    default_environment: str = "",
) -> Context:
    walrus_client = WalrusClient(
        walrus_url,
        walrus_api_key,
        verify=False,
    )
    if default_project != "" and default_environment != "":
        project = walrus_client.get_project(default_project)
        environment = walrus_client.get_environment(
            default_project, default_environment
        )
    else:
        # Get the first project and environment if not specified.
        projects = walrus_client.list_projects()
        if projects is None or len(projects) == 0:
            raise Exception("No available project. A project is required.")
        project = projects[0]
        environments = walrus_client.list_environments(project.get("id"))
        if environments is None or len(environments) == 0:
            raise Exception(
                "No aviailable environment. An environment is required."
            )
        environment = environments[0]

    global GLOBAL_CONTEXT
    GLOBAL_CONTEXT = Context(
        project_id=project.get("id"),
        project_name=project.get("name"),
        environment_id=environment.get("id"),
        environment_name=environment.get("name"),
    )


def update_context(context):
    global GLOBAL_CONTEXT
    if (
        context.get("project_id") is not None
        and context.get("project_name") != ""
    ):
        GLOBAL_CONTEXT.project_id = context.get("project_id")
        GLOBAL_CONTEXT.project_name = context.get("project_name")
    if (
        context.get("environment_id") is not None
        and context.get("environment_name") != ""
    ):
        GLOBAL_CONTEXT.environment_id = context.get("environment_id")
        GLOBAL_CONTEXT.environment_name = context.get("environment_name")

walrus/context.py
seal-io/appilot
import urllib3
from walrus import context
from walrus.tools.general.tools import BrowseURLTool
from walrus.tools.manage_context.tool import (
    ChangeContextTool,
    CurrentContextTool,
)
from walrus.tools.manage_environment.tool import (
    CloneEnvironmentTool,
    DeleteEnvironmentsTool,
    GetEnvironmentDependencyGraphTool,
    ListEnvironmentsTool,
)
from walrus.tools.manage_project.tool import ListProjectsTool
from walrus.tools.manage_service.tool import (
    ConstructServiceToCreateTool,
    ConstructServiceToUpdateTool,
    CreateServiceTool,
    DeleteServicesTool,
    GetServiceAccessEndpointsTool,
    GetServiceDependencyGraphTool,
    GetServiceResourceLogsReturnDirectTool,
    GetServiceResourceLogsTool,
    GetServicesTool,
    InformServiceReadyTool,
    ListServiceResourcesTool,
    ListServicesInAllEnvironmentsTool,
    ListServicesTool,
    UpdateServiceTool,
    WatchServicesTool,
)
from walrus.tools.manage_template.tool import (
    GetTemplateSchemaTool,
    MatchTemplateTool,
)
from walrus.client import WalrusClient
from langchain.schema.language_model import BaseLanguageModel
from utils import utils


class WalrusToolKit:
    """Walrus toolkit."""

    walrus_client: WalrusClient
    llm: BaseLanguageModel

    def __init__(self, llm: BaseLanguageModel):
        self.llm = llm
        self.init_client()

    def init_client(self):
        walrus_api_key = utils.get_env("WALRUS_API_KEY")
        walrus_url = utils.get_env("WALRUS_URL")
        walrus_default_project = utils.get_env("WALRUS_DEFAULT_PROJECT")
        walrus_default_environment = utils.get_env(
            "WALRUS_DEFAULT_ENVIRONMENT"
        )
        walrus_skip_tls_verify = utils.get_env_bool(
            "WALRUS_SKIP_TLS_VERIFY", False
        )
        if walrus_skip_tls_verify:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

        if not walrus_url:
            raise Exception("WALRUS_URL is not set")
        if not walrus_api_key:
            raise Exception("WALRUS_API_KEY is not set")

        self.walrus_client = WalrusClient(
            walrus_url,
            walrus_api_key,
            verify=(not walrus_skip_tls_verify),
        )
        context.set_default(
            walrus_url=walrus_url,
            walrus_api_key=walrus_api_key,
            default_project=walrus_default_project,
            default_environment=walrus_default_environment,
        )

    def get_tools(self):
        walrus_client = self.walrus_client
        llm = self.llm
        tools = [
            CurrentContextTool(),
            ChangeContextTool(walrus_client=walrus_client),
            ListProjectsTool(walrus_client=walrus_client),
            ListEnvironmentsTool(walrus_client=walrus_client),
            DeleteEnvironmentsTool(walrus_client=walrus_client),
            CloneEnvironmentTool(walrus_client=walrus_client),
            GetEnvironmentDependencyGraphTool(
                walrus_client=walrus_client, return_direct=True
            ),
            MatchTemplateTool(llm=llm, walrus_client=walrus_client),
            GetTemplateSchemaTool(walrus_client=walrus_client),
            ConstructServiceToCreateTool(llm=llm, walrus_client=walrus_client),
            ConstructServiceToUpdateTool(llm=llm, walrus_client=walrus_client),
            GetServicesTool(walrus_client=walrus_client),
            ListServicesTool(walrus_client=walrus_client),
            WatchServicesTool(walrus_client=walrus_client, return_direct=True),
            InformServiceReadyTool(
                walrus_client=walrus_client, return_direct=True
            ),
            ListServicesInAllEnvironmentsTool(walrus_client=walrus_client),
            CreateServiceTool(walrus_client=walrus_client),
            UpdateServiceTool(walrus_client=walrus_client),
            DeleteServicesTool(walrus_client=walrus_client),
            ListServiceResourcesTool(walrus_client=walrus_client),
            GetServiceResourceLogsTool(walrus_client=walrus_client),
            GetServiceResourceLogsReturnDirectTool(
                walrus_client=walrus_client, return_direct=True
            ),
            GetServiceAccessEndpointsTool(walrus_client=walrus_client),
            BrowseURLTool(),
            GetServiceDependencyGraphTool(walrus_client=walrus_client),
        ]
        return tools

walrus/toolkit.py
seal-io/appilot
from typing import Any
from langchain.agents.tools import BaseTool
from walrus.client import WalrusClient


class WalrusTool(BaseTool):
    """Tool to interacte with Walrus APIs."""

    walrus_client: WalrusClient

walrus/tools/base/tools.py
seal-io/appilot
"""General Tools."""

walrus/tools/general/__init__.py
seal-io/appilot
from langchain.agents.tools import BaseTool
import webbrowser


class BrowseURLTool(BaseTool):
    """Tool to access a URL in browser."""

    name = "open_url_in_browser"
    description = "Open a URL in browser. Input is a URL."

    def _run(self, url: str) -> str:
        webbrowser.open(url)
        return "The URL is opened in browser."

walrus/tools/general/tools.py
seal-io/appilot
"""Tool for managing Walrus project and environment context."""

walrus/tools/manage_context/__init__.py
seal-io/appilot
import json
from langchain.agents.tools import BaseTool
from walrus.client import WalrusClient
from walrus import context as walrus_context


class CurrentContextTool(BaseTool):
    """Tool to get current project and environment context."""

    name = "current_context"
    description = "Get current project and environment context."

    def _run(self, query: str) -> str:
        return json.dumps(walrus_context.GLOBAL_CONTEXT.__dict__)


class ChangeContextTool(BaseTool):
    """Tool to change project and environment context."""

    name = "change_context"
    description = (
        "Change project and environment context."
        "Input should be a json string with 2 keys: "
        "project_name, environment_name."
        "If users did not specify any of the two keys, leave it empty."
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            context = json.loads(text)
        except Exception as e:
            return e

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        if "project_name" in context and context["project_name"] != "":
            try:
                project = self.walrus_client.get_project(
                    context["project_name"]
                )
            except Exception as e:
                return e
            project_id = project["id"]
            context["project_id"] = project_id

        if "environment_name" in context and context["environment_name"] != "":
            try:
                environment = self.walrus_client.get_environment(
                    project_id, context["environment_name"]
                )
            except Exception as e:
                return e
            context["environment_id"] = environment["id"]

        walrus_context.update_context(context)

walrus/tools/manage_context/tool.py
seal-io/appilot
"""Tool for managing environments."""

walrus/tools/manage_environment/__init__.py
seal-io/appilot
import json
import os
from langchain.agents.tools import BaseTool
from i18n import text
from tools.base.tools import RequireApprovalTool
import pydot
from PIL import Image

from walrus.client import WalrusClient
from walrus import context as walrus_context


class ListEnvironmentsTool(BaseTool):
    """Tool to list environments."""

    name = "list_environments"
    description = (
        "List environments of a project."
        "Input should be a project id or an empty string indicating current project in the context."
    )
    walrus_client: WalrusClient

    def _run(self, project_id: str) -> str:
        if project_id == "":
            project_id = walrus_context.GLOBAL_CONTEXT.project_id
        try:
            environments = self.walrus_client.list_environments(project_id)
        except Exception as e:
            return e
        if environments is not None and len(environments) > 0:
            return json.dumps(environments)
        return "No environments found."


class DeleteEnvironmentsTool(RequireApprovalTool):
    """Tool to delete environments."""

    name = "delete_environments"
    description = 'Delete one or multiple environments. Input should be a list of object, each object contains 2 keys, "name" and "id" of an environment.'
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            environments = json.loads(text)
        except Exception as e:
            raise e

        ids = [env["id"] for env in environments if "id" in env]
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        try:
            self.walrus_client.delete_environments(project_id, ids)
        except Exception as e:
            return e

        return "Deletion started."


class GetEnvironmentDependencyGraphTool(BaseTool):
    """Tool to get environment dependency graph."""

    name = "get_environment_dependency_graph"
    description = "Get dependency graph of an environment. Input should be name or id of an environment."
    walrus_client: WalrusClient

    def show_graph(self, graph_data: dict):
        node_shape_map = {
            "Service": "box",
            "ServiceResourceGroup": "ellipse",
            "ServiceResource": "ellipse",
        }
        edge_style_map = {
            "Composition": "composition",
            "Realization": "dotted",
            "Dependency": "dashed",
        }

        node_height = 1
        graph = pydot.Dot(graph_type="digraph")
        node_ids = []
        for vertex in graph_data["vertices"]:
            label_suffix = ""
            if vertex["kind"] == "Service":
                label_suffix = "<br/>service"
            else:
                label_suffix = f"<br/>{vertex['extensions']['type']}"

            graph.add_node(
                pydot.Node(
                    name=vertex["id"],
                    label=f"<{vertex['name']}{label_suffix}>",
                    shape=node_shape_map[vertex["kind"]],
                    height=node_height,
                )
            )
            node_ids.append(vertex["id"])

        edge_ids = []
        for edge in graph_data["edges"]:
            # FIXME: This is a hack to avoid edges that are not in the graph. resolve in API.
            if (
                edge.get("start").get("id") not in node_ids
                or edge.get("end").get("id") not in node_ids
            ):
                continue
            # FIXME: This is a hack to avoid duplicate edges. resolve in API.
            edge_id = edge.get("start").get("id") + edge.get("end").get("id")
            if edge_id in edge_ids:
                continue

            graph.add_edge(
                pydot.Edge(
                    edge.get("start").get("id"),
                    edge.get("end").get("id"),
                    style=edge_style_map[edge.get("type")],
                )
            )
            edge_ids.append(edge_id)

        output_directory = "/tmp/appilot"
        if not os.path.exists(output_directory):
            os.makedirs(output_directory)
        image_path = os.path.join(output_directory, "dependency_graph.png")
        graph.write_png(image_path)
        image = Image.open(image_path)
        image.show()

    def _run(self, environment: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        if environment is None or environment == "":
            environment = walrus_context.GLOBAL_CONTEXT.environment_id

        graph_data = self.walrus_client.get_environment_graph(
            project_id, environment
        )
        self.show_graph(graph_data)

        return text.get("show_graph_message")


class CloneEnvironmentTool(RequireApprovalTool):
    """Tool to clone an environment."""

    name = "clone_environment"
    description = (
        "Clone an environment to a new one. "
        'Input should be a json string with two keys: "original_environment_name" and "target_environment_name".'
    )

    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            data = json.loads(text)
        except Exception as e:
            return e

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        original_environment_name = data.get("original_environment_name")
        target_environment_name = data.get("target_environment_name")

        try:
            environment = self.walrus_client.get_environment(
                project_id, original_environment_name
            )
            services = self.walrus_client.list_services(
                project_id, original_environment_name
            )
            environment["name"] = target_environment_name
            environment["services"] = services

            self.walrus_client.create_environment(project_id, environment)
        except Exception as e:
            return e

        return "Successfully cloned."

walrus/tools/manage_environment/tool.py
seal-io/appilot
"""Tool for managing projects."""

walrus/tools/manage_project/__init__.py
seal-io/appilot
import json
from langchain.agents.tools import BaseTool
from walrus.client import WalrusClient


class ListProjectsTool(BaseTool):
    """Tool to list projects."""

    name = "list_projects"
    description = "List projects."
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        try:
            projects = self.walrus_client.list_projects()
        except Exception as e:
            return e
        return json.dumps(projects)

walrus/tools/manage_project/tool.py
seal-io/appilot
"""Tools for managing services."""

walrus/tools/manage_service/__init__.py
seal-io/appilot
CONSTRUCT_SERVICE_TO_CREATE_PROMPT = """
You are a planner that constructs the expected service object given a user query describing a deployment task.
For your reference, you will be provided existing services and related template version if there's any in the environment. Choose a template and fill in the input variables for the service.

You should:
1) evaluate whether the service object can be constructed according to the user query. If no, say why.
2) if yes, output in the following format:
3) for sensitive attributes, reference existing variables in the format `${{var.variable_name}}`.

CONSTRUCTED SERVICE: <SERVICE_OBJECT_IN_ONE_LINE_JSON>

Strictly follow the above output format, do not add extra explanation or words.
Service id is needed in upgrade cases.
Environment info is needed in any cases.
Service status is not needed in any cases.
Do not miss the environment info in the constructed service object.
The output will be passed to an API controller that can format it into web requests and return the responses.

Example:

CONSTRUCTED SERVICE: {{"name":"example","template":{{"name":"webservice","version":"0.0.4"}},"environment":{{"id":"1234567"}},"attributes":{{"image":"nginx","ports":[80],"request_cpu":"0.1","request_memory":"128Mi"}}}}

----

Context: {context}
User query: {query}

EXISTING SERVICES:
{existing_services}

RELATED TEMPLATE:
{related_template}

"""


CONSTRUCT_SERVICE_TO_UPDATE_PROMPT = """
You are a planner that constructs the expected service object given a user query describing an upgrade task.
For your reference, you will be provided the service about to upgrade and related template version if there's any in the environment. Choose a template and fill in the input variables for the service.

You should:
1) evaluate whether the service object can be constructed according to the user query. If no, say why.
2) if yes, output in the following format:

CONSTRUCTED SERVICE: <SERVICE_OBJECT_IN_ONE_LINE_JSON>

Strictly follow the above output format, do not add extra explanation or words.
Service id is required.
Environment info is required.
Service status is not needed.
The output will be used in the update API call of the service.

Example:

CONSTRUCTED SERVICE: {{"name":"example","template":{{"name":"webservice","version":"0.0.4"}},"environment":{{"id":"1234567"}},"attributes":{{"image":"nginx","ports":[80],"request_cpu":"0.1","request_memory":"128Mi"}}}}

----

Context: {context}
User query: {query}

CURRENT SERVICE:
{current_service}

RELATED TEMPLATE:
{related_template}

"""

walrus/tools/manage_service/prompt.py
seal-io/appilot
import json
import threading
import time

from utils import utils
from i18n import text
from walrus.client import WalrusClient
from langchain.agents.tools import BaseTool
from langchain import LLMChain
from langchain.prompts import PromptTemplate
from langchain.schema.language_model import BaseLanguageModel
from tools.base.tools import RequireApprovalTool
from walrus import context as walrus_context
from walrus.tools.manage_service.prompt import (
    CONSTRUCT_SERVICE_TO_CREATE_PROMPT,
    CONSTRUCT_SERVICE_TO_UPDATE_PROMPT,
)


class ListServicesTool(BaseTool):
    """Tool to list services."""

    name = "list_services"
    description = "List services in current environment."
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            services = self.walrus_client.list_services(
                project_id, environment_id
            )
        except Exception as e:
            return e

        if services is not None and len(services) > 0:
            return json.dumps(services)

        return "No services found."


class WatchServicesTool(BaseTool):
    """Tool to watch services."""

    name = "watch_services"
    description = "Watch service changes in current environment."
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id

        try:
            self.walrus_client.watch_services(project_id, environment_id)
        except KeyboardInterrupt:
            # Ctrl+C detected. Stopping the request.
            print("")

        return text.get("watch_service_ending")


class InformServiceReadyTool(BaseTool):
    """Tool to inform user when service becomes ready."""

    name = "inform_service_ready"
    description = "Inform user when a service becomes ready. Input should be name or id of a service."
    walrus_client: WalrusClient

    def watch_service_ready(self, input: str):
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        start_time = time.time()
        timeout = 600
        while True:
            time.sleep(3)
            if time.time() - start_time > timeout:
                break
            service = self.walrus_client.get_service_by_name(
                project_id, environment_id, input
            )
            if service.get("status").get("summaryStatus") == "Ready":
                utils.print_ai_inform(
                    text.get("service_ready_message").format(input)
                )
                break

    def _run(self, input: str) -> str:
        threading.Thread(
            target=self.watch_service_ready, args=(input,)
        ).start()
        return text.get("inform_ready_start")


class ListServicesInAllEnvironmentsTool(BaseTool):
    """Tool to list services in all environments."""

    name = "list_services_in_all_environments"
    description = "List services in all environments of current project."
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        try:
            services = self.walrus_client.list_services_in_all_environments(
                project_id
            )
        except Exception as e:
            return e

        if services is not None and len(services) > 0:
            return json.dumps(services)

        return "No services found."


class GetServicesTool(BaseTool):
    """Tool to get a service."""

    name = "get_service"
    description = "Get a service object. Input should be a service name."
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            service = self.walrus_client.get_service_by_name(
                project_id, environment_id, query
            )
        except Exception as e:
            return e

        return json.dumps(service)


class CreateServiceTool(RequireApprovalTool):
    """Tool to create a service."""

    name = "create_service"
    description = (
        "Create a service."
        "Input should be a service object in json format."
        'Output a json string with 2 keys, "id" and "name" of the service.'
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            service = json.loads(text)
        except Exception as e:
            raise e

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            self.walrus_client.create_service(
                project_id, environment_id, service
            )
        except Exception as e:
            return e

        return "Successfully created."


class UpdateServiceTool(RequireApprovalTool):
    """Tool to update a service."""

    name = "update_service"
    description = (
        "Update a service."
        "Input should be a service object in json format."
        'Output a json string with 2 keys, "id" and "name" of the service.'
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            self.walrus_client.update_service(project_id, environment_id, text)
        except Exception as e:
            return e

        return "Successfully updated."


class DeleteServicesTool(RequireApprovalTool):
    """Tool to delete one or multiple services."""

    name = "delete_services"
    description = 'Delete one or multiple services. Input should be a list of object, each object contains 2 keys, "name" and "id" of a service.'
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        try:
            services = json.loads(query)
        except Exception as e:
            raise e

        ids = [service["id"] for service in services if "id" in service]
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            self.walrus_client.delete_services(project_id, environment_id, ids)
        except Exception as e:
            return e

        return "Deletion started."


class GetServiceAccessEndpointsTool(BaseTool):
    """Tool to get access endpoints of a service."""

    name = "get_service_access_endpoints"
    description = (
        "Get access endpoints of a service."
        "Input should be id of a service."
        "Output service access endpoints."
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            endpoints = self.walrus_client.get_service_access_endpoints(
                project_id, environment_id, text
            )
        except Exception as e:
            return e

        return endpoints


class ListServiceResourcesTool(BaseTool):
    """Tool to get resources of a service."""

    name = "get_service_resources"
    description = (
        "Get resources of a service. "
        "Helpful to know what resources a service consists of, what status they are in. what keys they have."
        "Input should be id of a service. "
        "Output resource objects in json format."
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            resources = self.walrus_client.list_service_resources(
                project_id, environment_id, text
            )
        except Exception as e:
            return e

        return resources


class GetServiceDependencyGraphTool(BaseTool):
    """Tool to get service dependency graph."""

    name = "get_service_dependency_graph"
    description = (
        "Get dependency graph of a service. Input should be a service id."
        "Output is a json data wrapped in triple backticks, representing the dependency graph."
        "You can directly return the output to the user. No need to reformat. "
        "UI can use this data to render the graph."
    )
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        data = {
            "project_id": walrus_context.GLOBAL_CONTEXT.project_id,
            "service_id": text,
        }
        return f"```service_resource_graph\n{data}\n```"


class GetServiceResourceLogsTool(BaseTool):
    """Tool to get logs of a service resource."""

    name = "get_service_resource_logs_for_diagnose"
    description = (
        "Get logs of a service resource. Use when you need to diagnose service error with logs."
        "Before using this tool, you should get keys of the resource first."
        'Input should be a json with 4 keys: "service_id", "service_resource_id", "key", "line_number".'
        '"key" is identity of a service resource\'s component. You can get available keys by listing service resources. '
        '"line_number" is the number of lines of logs to get. defaults to 100 if user does not specify. '
        "Output is log text."
    )
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        try:
            input = json.loads(query)
        except Exception as e:
            raise e
        service_id = input.get("service_id")
        service_resource_id = input.get("service_resource_id")
        key = input.get("key")
        line_number = input.get("line_number", 100)

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            log = self.walrus_client.get_service_resource_logs(
                project_id,
                environment_id,
                service_id,
                service_resource_id,
                key,
                line_number,
            )
        except Exception as e:
            return e

        prefix = text.get("resource_log_prefix")
        return f"{prefix}\n```{log}```"


class GetServiceResourceLogsReturnDirectTool(BaseTool):
    """Tool to get logs of a service resource."""

    name = "get_service_resource_logs_return_direct"
    description = (
        "Get logs of a service resource. The logs will be shown to users directly. Useful when users want to see logs."
        "Before using this tool, you should get keys of the resource first."
        'Input should be a json with 4 keys: "service_id", "service_resource_id", "key", "line_number".'
        '"key" is identity of a service resource\'s component. You can get available keys by listing service resources. '
        '"line_number" is the number of lines of logs to get. defaults to 100 if user does not specify. '
        "Output is log text."
    )
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        try:
            input = json.loads(query)
        except Exception as e:
            raise e
        service_id = input.get("service_id")
        service_resource_id = input.get("service_resource_id")
        key = input.get("key")
        line_number = input.get("line_number", 100)

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        try:
            log = self.walrus_client.get_service_resource_logs(
                project_id,
                environment_id,
                service_id,
                service_resource_id,
                key,
                line_number,
            )
        except Exception as e:
            return e

        prefix = text.get("resource_log_prefix")
        return f"{prefix}\n```{log}```"


class ConstructServiceToCreateTool(BaseTool):
    """Construct a service for deployment in Walrus system."""

    name = "construct_service_to_create"
    description = (
        "Construct a service for creation in Walrus system."
        'Input to the tool should be a json with 3 keys: "user_query" and "related_template_name".'
        'The value of "user_query" should be the description of a deployment task.'
        'The value of "related_template_name" should be name of a template related to the deployment task.'
        "The output is a service object in json. It will be used in the creation of a service."
    )
    llm: BaseLanguageModel
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            data = json.loads(text)
        except Exception as e:
            raise e

        query = data.get("user_query")
        template_name = data.get("related_template_name")

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        existing_services = self.walrus_client.list_services(
            project_id, environment_id
        )
        related_template = self.walrus_client.get_template_version(
            template_name
        )

        prompt = PromptTemplate(
            template=CONSTRUCT_SERVICE_TO_CREATE_PROMPT,
            input_variables=["query"],
            partial_variables={
                "context": json.dumps(walrus_context.GLOBAL_CONTEXT.__dict__),
                "existing_services": json.dumps(existing_services),
                "related_template": json.dumps(related_template),
            },
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        return chain.run(json.dumps(query)).strip()


class ConstructServiceToUpdateTool(BaseTool):
    """Construct a service for upgrade in Walrus system."""

    name = "construct_service_to_update"
    description = (
        "Construct a service for update in Walrus system."
        'Input to the tool should be a json with 3 keys: "user_query", "service_name" and "related_template_name".'
        'The value of "user_query" should be the description of a deployment task.'
        'The value of "service_name" should be name of the service about to update.'
        'The value of "related_template_name" should be name of a template related to the deployment task.'
        "The output is a service object in json. It will be used in the update of a service."
    )
    llm: BaseLanguageModel
    walrus_client: WalrusClient

    def _run(self, text: str) -> str:
        try:
            data = json.loads(text)
        except Exception as e:
            raise e

        query = data.get("user_query")
        service_name = data.get("service_name")
        template_name = data.get("related_template_name")

        project_id = walrus_context.GLOBAL_CONTEXT.project_id
        environment_id = walrus_context.GLOBAL_CONTEXT.environment_id
        related_template = self.walrus_client.get_template_version(
            template_name
        )
        current_service = self.walrus_client.get_service_by_name(
            project_id=project_id,
            environment_id=environment_id,
            service_name=service_name,
        )

        prompt = PromptTemplate(
            template=CONSTRUCT_SERVICE_TO_UPDATE_PROMPT,
            input_variables=["query"],
            partial_variables={
                "context": json.dumps(walrus_context.GLOBAL_CONTEXT.__dict__),
                "current_service": json.dumps(current_service),
                "related_template": json.dumps(related_template),
            },
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        return chain.run(json.dumps(query)).strip()

walrus/tools/manage_service/tool.py
seal-io/appilot
"""Tools for managing templates."""

walrus/tools/manage_template/__init__.py
seal-io/appilot
FIND_TEMPLATE_PROMPT = """
Templates are predefined configuration to create a particular type of service.

You will be provided existing templates and a user query describing a deployment task. 
Find a template that most likely can be used to accomplish the user query.
If you don't find any, say why.

Output the matched template name in quoted string.

TEMPLATES:
{templates}

User query: {query}
Output:"""

walrus/tools/manage_template/prompt.py
seal-io/appilot
import json
from langchain import LLMChain

from langchain.agents.tools import BaseTool
from langchain.prompts import PromptTemplate
from langchain.schema.language_model import BaseLanguageModel
from walrus.tools.manage_template.prompt import FIND_TEMPLATE_PROMPT
from walrus.client import WalrusClient


class MatchTemplateTool(BaseTool):
    """Find matching template useful for a deployment task.
    Input should be a deployment task.
    Output matching template name.
    """

    name = "find_matching_template"
    description = (
        "Find a matching template for a deploy/upgrade task."
        "Input should be description of the task. For upgrade task, include previous template info."
        "Output matching template name, or None when no matching template found."
    )
    walrus_client: WalrusClient
    llm: BaseLanguageModel

    def _run(self, query: str) -> str:
        try:
            templates = self.walrus_client.list_templates()
        except Exception as e:
            return e

        prompt = PromptTemplate(
            template=FIND_TEMPLATE_PROMPT,
            input_variables=["query"],
            partial_variables={
                "templates": json.dumps(templates),
            },
        )
        chain = LLMChain(llm=self.llm, prompt=prompt)
        return chain.run(query)


class GetTemplateSchemaTool(BaseTool):
    """Tool to get template version and schema given template name."""

    name = "get_template_schema"
    description = (
        "Get template version and schema given template name."
        "Input should be a template name."
        "Output template schema."
    )
    walrus_client: WalrusClient

    def _run(self, query: str) -> str:
        try:
            template_version = self.walrus_client.get_template_version(query)
        except Exception as e:
            return e

        return template_version

walrus/tools/manage_template/tool.py
seal-io/appilot