Spaces:
Paused
Paused
File size: 2,856 Bytes
8d1819a |
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
|