File size: 2,856 Bytes
de76de9
f6f0dbf
e2b7e2b
f6f0dbf
21be64e
e06dee8
f6f0dbf
 
 
 
e06dee8
 
 
 
 
 
537c89a
 
 
317ffd2
321abe1
e06dee8
317ffd2
9112b0b
 
760ba6e
 
321abe1
 
e06dee8
 
f6f0dbf
4058185
537c89a
215dbf1
f6f0dbf
4058185
537c89a
215dbf1
4058185
e06dee8
317ffd2
 
 
 
760ba6e
 
 
e06dee8
 
 
760ba6e
321abe1
760ba6e
 
321abe1
 
 
 
 
7f9e34a
e06dee8
7f9e34a
321abe1
21be64e
 
321abe1
 
 
e06dee8
 
 
 
7f9e34a
 
 
e06dee8
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
from typing import Any
from python.helpers.extension import Extension
from python.helpers.mcp_handler import MCPConfig
from agent import Agent, LoopData
from python.helpers.settings import get_settings
from python.helpers import projects


class SystemPrompt(Extension):

    async def execute(
        self,
        system_prompt: list[str] = [],
        loop_data: LoopData = LoopData(),
        **kwargs: Any
    ):
        # append main system prompt and tools
        main = get_main_prompt(self.agent)
        tools = get_tools_prompt(self.agent)
        mcp_tools = get_mcp_tools_prompt(self.agent)
        secrets_prompt = get_secrets_prompt(self.agent)
        project_prompt = get_project_prompt(self.agent)

        system_prompt.append(main)
        system_prompt.append(tools)
        if mcp_tools:
            system_prompt.append(mcp_tools)
        if secrets_prompt:
            system_prompt.append(secrets_prompt)
        if project_prompt:
            system_prompt.append(project_prompt)


def get_main_prompt(agent: Agent):
    return agent.read_prompt("agent.system.main.md")


def get_tools_prompt(agent: Agent):
    prompt = agent.read_prompt("agent.system.tools.md")
    if agent.config.chat_model.vision:
        prompt += "\n\n" + agent.read_prompt("agent.system.tools_vision.md")
    return prompt


def get_mcp_tools_prompt(agent: Agent):
    mcp_config = MCPConfig.get_instance()
    if mcp_config.servers:
        pre_progress = agent.context.log.progress
        agent.context.log.set_progress(
            "Collecting MCP tools"
        )  # MCP might be initializing, better inform via progress bar
        tools = MCPConfig.get_instance().get_tools_prompt()
        agent.context.log.set_progress(pre_progress)  # return original progress
        return tools
    return ""


def get_secrets_prompt(agent: Agent):
    try:
        # Use lazy import to avoid circular dependencies
        from python.helpers.secrets import get_secrets_manager

        secrets_manager = get_secrets_manager(agent.context)
        secrets = secrets_manager.get_secrets_for_prompt()
        vars = get_settings()["variables"]
        return agent.read_prompt("agent.system.secrets.md", secrets=secrets, vars=vars)
    except Exception as e:
        # If secrets module is not available or has issues, return empty string
        return ""


def get_project_prompt(agent: Agent):
    result = agent.read_prompt("agent.system.projects.main.md")
    project_name = agent.context.get_data(projects.CONTEXT_DATA_KEY_PROJECT)
    if project_name:
        project_vars = projects.build_system_prompt_vars(project_name)
        result += "\n\n" + agent.read_prompt(
            "agent.system.projects.active.md", **project_vars
        )
    else:
        result += "\n\n" + agent.read_prompt("agent.system.projects.inactive.md")
    return result