Agent AI krok po kroku - od Pythona po Ansible

1

git clone https://github.com/sysadmin-info/ai-agent.git

1

uvicorn index:app --host 0.0.0.0 --port 3000

1
2

alias activate-agent="source $HOME/agent/venv/bin/activate"
alias activate-aidevs="source $HOME/aidevs/agent/venv/bin/activate"

1

source ~/.bashrc

  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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196

import logging
from quart import Quart, request, jsonify
from lib.prompts import Prompts
from lib.ai import AnthropicCompletion
from typing import Dict, Any
from dotenv import load_dotenv
import os
import json
import datetime

# Załaduj zmienne środowiskowe
load_dotenv()

# Inicjalizacja aplikacji Quart
app = Quart(__name__)

# Konfiguracja loggera
class SensitiveDataFilter(logging.Filter):
    """
    Filtr logowania do ukrywania danych wrażliwych, takich jak klucz API.
    """
    def filter(self, record):
        if record.msg and isinstance(record.msg, str):
            record.msg = record.msg.replace(os.getenv("ANTHROPIC_API_KEY", ""), "[REDACTED]")
        return True

logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("AIAgentLogger")
logger.addFilter(SensitiveDataFilter())


class AIAgent:
    def __init__(self, api_key: str):
        self.completion_client = AnthropicCompletion(api_key)
        self.state = {
            "currentStage": "init",
            "currentStep": 1,
            "maxSteps": 15,
            "messages": [],
            "systemPrompt": "",
            "plan": "",
            "actionsTaken": [],
            "activeTool": {},
            "api_key": api_key
        }

    def _sanitize_state(self) -> Dict[str, Any]:
        """
        Usuwa wrażliwe dane (np. klucz API) przed logowaniem.
        """
        return {k: v for k, v in self.state.items() if k != "api_key"}

    async def final_answer(self) -> str:
        self.state["systemPrompt"] = Prompts.final_answer_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Sending final_answer request: {messages}")
        answer = await self.completion_client.completion(messages)
        parsed_answer = self._parse_response(answer, step="final_answer")
        self._log_to_markdown("result", "Final Answer", json.dumps(parsed_answer))
        return parsed_answer

    async def run(self, initial_message: str) -> str:
        self.state["messages"] = [{"role": "user", "content": initial_message}]
        logger.debug(f"Initial state: {self._sanitize_state()}")

        while self.state["currentStep"] <= self.state["maxSteps"]:
            try:
                self._log_request_start("plan")
                await self._plan()
                self._log_request_end("plan")

                self._log_request_start("decide")
                await self._decide()
                self._log_request_end("decide")

                if not self.state.get("activeTool", {}).get("tool"):
                    raise ValueError("Active tool is not defined or missing the 'tool' property in state.")

                if self.state["activeTool"]["tool"] == "final_answer":
                    return await self.final_answer()

                self._log_request_start("describe")
                await self._describe()
                self._log_request_end("describe")

                self._log_request_start("execute")
                await self._execute()
                self._log_request_end("execute")

                self._log_request_start("reflect")
                await self._reflect()
                self._log_request_end("reflect")

                self.state["currentStep"] += 1
            except Exception as e:
                logger.error(f"Error during step {self.state['currentStage']}: {str(e)}")
                raise

    async def _plan(self):
        self.state["currentStage"] = "plan"
        self.state["systemPrompt"] = Prompts.plan_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Plan request payload: {messages}")
        plan_response = await self.completion_client.completion(messages)
        self.state["plan"] = self._parse_response(plan_response, step="plan")

    async def _decide(self):
        self.state["currentStage"] = "decide"
        self.state["systemPrompt"] = Prompts.decide_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Decide request payload: {messages}")
        decision_response = await self.completion_client.completion(messages)
        try:
            self.state["activeTool"] = json.loads(self._parse_response(decision_response, step="decide"))
            logger.debug(f"Active tool decided: {self.state['activeTool']}")
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse decision JSON: {decision_response}")
            raise ValueError(f"Error parsing decision JSON: {decision_response}") from e

    async def _describe(self):
        self.state["currentStage"] = "describe"
        if not self.state.get("activeTool", {}).get("tool"):
            raise ValueError("Active tool is not defined or missing the 'tool' property in state.")

        self.state["systemPrompt"] = Prompts.describe_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Describe request payload: {messages}")
        describe_response = await self.completion_client.completion(messages)
        self.state["activeToolPayload"] = self._parse_response(describe_response, step="describe")

    async def _execute(self):
        self.state["currentStage"] = "execute"
        tool_name = self.state["activeTool"]["tool"]
        payload = self.state["activeToolPayload"]
        logger.debug(f"Executing tool: {tool_name} with payload: {payload}")
        result = f"Executed {tool_name} with payload {payload}"
        self.state["actionsTaken"].append({
            "name": tool_name,
            "payload": payload,
            "result": result,
            "reflection": ""
        })

    async def _reflect(self):
        self.state["currentStage"] = "reflect"
        self.state["systemPrompt"] = Prompts.reflection_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Reflect request payload: {messages}")
        reflection_response = await self.completion_client.completion(messages)
        self.state["actionsTaken"][-1]["reflection"] = self._parse_response(reflection_response, step="reflect")

    def _parse_response(self, response: Dict[str, Any], step: str) -> str:
        logger.debug(f"Raw response for step {step}: {response}")
        try:
            if "completion" in response:
                return response["completion"]
            elif "content" in response:
                content_list = response.get("content", [])
                if content_list:
                    return content_list[0].get("text", "")
            raise ValueError("Invalid response format")
        except Exception as e:
            logger.error(f"Failed to parse response: {str(e)}")
            raise ValueError(f"Error parsing API response for step {step}: {response}")

    def _log_request_start(self, step):
        logger.debug(f"Step '{step}' started at {datetime.datetime.now()}")

    def _log_request_end(self, step):
        logger.debug(f"Step '{step}' ended at {datetime.datetime.now()}")

    def _log_to_markdown(self, log_type: str, header: str, content: str):
        with open("log.md", "a") as f:
            f.write(f"## {header}\n\n{content}\n\n")


@app.route("/", methods=["POST"])
async def process_request():
    data = await request.get_json()
    initial_message = data.get("messages", "")

    logger.debug(f"Incoming message: {initial_message}")

    api_key = os.getenv("ANTHROPIC_API_KEY")
    if not api_key:
        raise ValueError("API key is missing in environment variables or .env file.")

    agent = AIAgent(api_key)

    try:
        result = await agent.run(initial_message)
        logger.debug(f"Final sanitized state: {agent._sanitize_state()}")
        return jsonify({"response": result})
    except Exception as e:
        logger.error(f"Exception during agent execution: {str(e)}")
        return jsonify({"error": str(e)}), 500

 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

from typing import List, Optional, TypedDict, Literal, Any

# Define the stages that the agent can go through
Stage = Literal['init', 'plan', 'decide', 'describe', 'reflect', 'execute', 'final']

class ITool(TypedDict):
    """
    Represents a tool that the AI agent can use, with a name, instruction, and description.
    """
    name: str
    instruction: str
    description: str

class IAction(TypedDict):
    """
    Represents an action taken by the AI agent, with fields for the action name, payload, result, reflection, and the tool used.
    """
    name: str
    payload: str
    result: str
    reflection: str
    tool: str

class IState(TypedDict, total=False):
    """
    Represents the state of the AI agent, including system prompts, messages, the current stage and step,
    and actions taken so far.
    """
    systemPrompt: str                      # Current system prompt
    messages: List[dict]                   # All messages in the conversation
    currentStage: Stage                    # Stage on which the system prompt depends
    currentStep: int                       # Current step in the agent's operation
    maxSteps: int                          # Maximum number of steps allowed
    activeTool: Optional[ITool]            # The tool currently being used by the agent
    activeToolPayload: Optional[Any]       # Payload for the active tool
    plan: str                              # Current plan of action
    actionsTaken: List[IAction]            # List of actions taken so far

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

 # task_manager.py
 class TaskManager:
     def __init__(self):
         pass

     def get_task(self, task_name):
         # Placeholder for retrieving a task based on task_name
         # Replace with actual logic
         tasks = {
             "example_task": {"function": self.example_task}
         }
         return tasks.get(task_name)

     def example_task(self):
         # Example task logic
         return "Task executed successfully"

 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

 # ssh_manager.py
 import asyncssh
 import asyncio

 class AsyncSSHManager:
     def __init__(self, hostname: str, username: str, password: str):
         self.hostname = hostname
         self.username = username
         self.password = password
         self.connection = None

     async def connect(self):
         """Establish an SSH connection."""
         self.connection = await asyncssh.connect(
             self.hostname,
             username=self.username,
             password=self.password
         )

     async def execute_command(self, command: str) -> str:
         """Execute a command over SSH and return the output."""
         if self.connection is None:
             raise ValueError("No active SSH connection")
         result = await self.connection.run(command, check=True)
         return result.stdout

     async def close_connection(self):
         """Close the SSH connection."""
         if self.connection:
             self.connection.close()
             await self.connection.wait_closed()
             self.connection = None

 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

import os
import httpx
from dotenv import load_dotenv
import time

# Załaduj zmienne środowiskowe
load_dotenv()

class AnthropicCompletion:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
        if not self.api_key:
            raise ValueError("API key not provided or incorrectly set in environment variables")

    async def completion(self, messages: list, model: str = "claude-3-5-sonnet-20241022", retries: int = 3, delay: int = 5) -> dict:
        url = "https://api.anthropic.com/v1/messages"
        headers = {
            "x-api-key": self.api_key,
            "Content-Type": "application/json",
            "anthropic-version": "2023-06-01"
        }
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": 1000,
            "temperature": 0.7
        }

        for attempt in range(retries):
            async with httpx.AsyncClient() as client:
                try:
                    response = await client.post(url, headers=headers, json=payload)
                    if response.status_code == 529:
                        time.sleep(delay)
                        continue
                    response.raise_for_status()
                    return response.json()
                except httpx.RequestError as e:
                    if attempt == retries - 1:
                        raise
                except httpx.HTTPStatusError as e:
                    if attempt == retries - 1:
                        raise

  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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

from typing import Dict

class Prompts:
    @staticmethod
    def tools_instruction() -> Dict[str, str]:
        return {
            "get_html_contents": (
                'Required payload: {"url": "URL that needs to be downloaded"} '
                'Response format: HTML content of the page.'
            ),
            "game_submit_form": (
                'Required payload: {"url": "URL to a file that will be passed to the game"}. '
                'Response format: The game\'s response after submitting the form.'
            ),
            "upload_text_file": (
                'Required payload: {"content": "Text content of the file", '
                '"file_name": "Name of the file (e.g., document.md)"} '
                'Response format: URL of the uploaded file.'
            ),
            "final_answer": (
                'Required payload: {"answer": "Your final answer"}. '
                'Response format: A direct response to the user.'
            ),
            "play_music": (
                'Required payload: JSON object with Spotify API details for actions like search, play, or playlist creation.'
            )
        }

    @staticmethod
    def available_tools() -> Dict[str, str]:
        return {
            "get_html_contents": "Fetch HTML content of a URL.",
            "upload_text_file": "Create and upload a text file.",
            "game_submit_form": "Submit a URL to the game.",
            "final_answer": "Provide the final response to the user.",
            "play_music": "Generate Spotify API JSON for playing or managing music."
        }

    @staticmethod
    def extract_user_query(state) -> str:
        try:
            return next(
                (msg["content"] for msg in state.get("messages", []) if msg.get("role") == "user"),
                "No specific query provided."
            )
        except Exception:
            return "No specific query provided."

    @staticmethod
    def plan_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Analyze the user's query and decide whether to provide an immediate answer or develop a detailed plan.
</main_objective>

<rules>
- If the query is straightforward (e.g., "How far is the Moon from Earth?"), prioritize addressing it directly.
- If the query requires multiple steps, use available tools to create an actionable plan.
- Always respond with clarity and avoid unnecessary complexity.
</rules>

<user_query>
{user_query}
</user_query>

<available_tools>
{Prompts.tools_instruction()}
</available_tools>
"""

    @staticmethod
    def decide_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Determine the next step based on the user's query and current context. Either select the appropriate tool to proceed or decide to provide the final answer.
</main_objective>

<rules>
- Be concise and provide a JSON response with the selected tool and reasoning.
- If the question is straightforward, move directly to the final answer.
- Always return a valid JSON string with the tool name.
- The JSON structure must include:
  {{
    "_thoughts": "Your internal reasoning",
    "tool": "precise name of the tool"
  }}
</rules>

<user_query>
{user_query}
</user_query>

<available_tools>
{Prompts.available_tools()}
</available_tools>

<current_plan>
Plan: {state['plan'] if state.get('plan') else 'No plan yet.'}
</current_plan>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def describe_prompt(state) -> str:
        if "activeTool" not in state or not state["activeTool"].get("tool"):
            raise ValueError("Active tool is not defined or missing the 'tool' property.")

        return f"""
<main_objective>
Provide the required details to execute the tool "{state['activeTool']['tool']}" based on the current state.
</main_objective>

<tool_details>
Tool Name: {state['activeTool']['tool']}
Tool Instructions: {state['activeTool'].get('instruction', 'No instructions available')}
</tool_details>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def reflection_prompt(state) -> str:
        return f"""
<main_objective>
Reflect on the last action performed and suggest improvements or adjustments to the plan if needed.
</main_objective>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def final_answer_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Provide the final answer to the user's query: "{user_query}".
</main_objective>

<rules>
- Directly answer the user's question in a clear and actionable manner.
- If the query is unclear, ask for clarification.
- Summarize key findings and insights.
</rules>

<current_plan>
{state['plan'] if state.get('plan') else 'No plan created.'}
</current_plan>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

  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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

import json
from lib.tools import browse, upload_file, play_music
from lib.prompts import Prompts
from lib.ai import AnthropicCompletion  # Use the AnthropicCompletion class
from typing import Dict, Any

def log_to_markdown(type: str, header: str, content: str):
    """
    Logs content to a markdown file.

    Parameters:
    - type (str): The type of log entry (e.g., 'basic', 'action', 'result').
    - header (str): The header of the log entry.
    - content (str): The content to log.
    """
    formatted_content = f"## {header}\n\n{content}\n\n" if type == "basic" else \
                        f"### {header}\n\n{content}\n\n" if type == "action" else \
                        f"#### {header}\n```\n{content}\n```\n\n"

    with open("log.md", "a") as f:
        f.write(formatted_content)

async def plan(state, anthropic_completion):
    """
    Generates a plan based on the current state.
    """
    state["currentStage"] = "plan"
    state["systemPrompt"] = Prompts.plan_prompt(state)
    state["plan"] = await anthropic_completion.completion(state["systemPrompt"])
    log_to_markdown("basic", "Planning", f"Current plan: {state['plan']}")

async def decide(state, anthropic_completion):
    """
    Decides the next tool to use based on the current state.
    """
    state["currentStage"] = "decide"
    state["systemPrompt"] = Prompts.decide_prompt(state)
    next_step = await anthropic_completion.completion(state["systemPrompt"])
    state["activeTool"] = {
        "name": next_step["tool"],
        "description": Prompts.available_tools().get(next_step["tool"]),
        "instruction": Prompts.tools_instruction().get(next_step["tool"])
    }
    log_to_markdown("action", "Decision", f"Next move: {json.dumps(next_step)}")

async def describe(state, anthropic_completion):
    """
    Generates a payload description for the active tool.
    """
    state["currentStage"] = "describe"
    state["systemPrompt"] = Prompts.describe_prompt(state)
    next_step = await anthropic_completion.completion(state["systemPrompt"])
    state["activeToolPayload"] = next_step
    log_to_markdown("action", "Description", f"Next step description: {json.dumps(next_step)}")

async def execute(state):
    """
    Asynchronously executes the active tool with the generated payload.
    """
    state["currentStage"] = "execute"
    if not state.get("activeTool"):
        raise ValueError("No active tool to execute")

    tool_name = state["activeTool"]["name"]
    payload = state.get("activeToolPayload", {})

    if tool_name == "get_html_contents":
        result = await browse(payload["url"])  # Async call to browse
    elif tool_name == "upload_text_file":
        result = await upload_file(payload)  # Async call to upload_file
    elif tool_name == "play_music":
        result = await play_music(payload)  # Async call to play_music
    else:
        result = f"Tool '{tool_name}' execution not defined."

    log_to_markdown("result", "Execution", f"Action result: {json.dumps(result)}")
    state["actionsTaken"].append({
        "name": tool_name,
        "payload": json.dumps(payload),
        "result": result,
        "reflection": ""
    })

async def reflect(state, anthropic_completion):
    """
    Reflects on the results of the last action.
    """
    state["currentStage"] = "reflect"
    state["systemPrompt"] = Prompts.reflection_prompt(state)
    reflection = await anthropic_completion.completion(state["systemPrompt"])
    state["actionsTaken"][-1]["reflection"] = reflection
    log_to_markdown("basic", "Reflection", reflection)

class AIAgent:
    def __init__(self, api_key: str):
        self.completion_client = AnthropicCompletion(api_key)
        self.state = {
            "currentStage": "init",
            "currentStep": 1,
            "maxSteps": 15,
            "messages": [],
            "systemPrompt": "",
            "plan": "",
            "actionsTaken": []
        }

    async def final_answer(self) -> str:
        """
        Generates the final answer based on the entire process.
        """
        self.state["systemPrompt"] = Prompts.final_answer_prompt(self.state)
        answer = await self.completion_client.completion(self.state["systemPrompt"])
        log_to_markdown("result", "Final Answer", json.dumps(answer))
        return answer

    async def run(self, initial_message: Dict[str, Any]) -> str:
        """
        Executes the agent's full process from planning to providing the final answer.

        Parameters:
        - initial_message (Dict[str, Any]): The initial message or prompt for the agent.

        Returns:
        - str: The final answer.
        """
        self.state["messages"] = [initial_message]

        while self.state["currentStep"] <= self.state["maxSteps"]:
            await plan(self.state, self.completion_client)
            await decide(self.state, self.completion_client)

            if self.state.get("activeTool", {}).get("name") == "final_answer":
                break

            await describe(self.state, self.completion_client)
            await execute(self.state)
            await reflect(self.state, self.completion_client)
            self.state["currentStep"] += 1

        return await self.final_answer()

  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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

import httpx
from markdownify import markdownify as md
import os
import re

async def browse(url: str) -> str:
    """
    Asynchronicznie pobiera zawartość HTML z podanego URL i konwertuje ją do Markdown.

    Parameters:
    - url (str): The URL to fetch content from.

    Returns:
    - str: The markdown content of the fetched HTML, or an error message if the request fails.
    """
    if url in ["https://aidevs.pl", "https://www.aidevs.pl"]:
        return "You can't browse the main website. Try another URL."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            response.raise_for_status()
            html_content = response.text

            # Extract script contents
            script_contents = ""
            script_tags = re.findall(r"<script\b[^>]*>([\s\S]*?)<\/script>", html_content)
            for i, script in enumerate(script_tags, 1):
                script_contents += f"\n\n--- Script {i} ---\n{script}"

            # Convert HTML to markdown using markdownify
            markdown_content = md(html_content)

            # Combine markdown content with script contents
            return f"{markdown_content}\n\n--- Script Contents ---{script_contents}"
    except httpx.RequestError as e:
        print("Error fetching URL:", e)
        return "Failed to fetch the URL, please try again."

async def upload_file(data: dict) -> str:
    """
    Asynchronicznie przesyła plik tekstowy na serwer i zwraca URL pliku.

    Parameters:
    - data (dict): Contains "content" and "file_name" keys.

    Returns:
    - str: The URL of the uploaded file or an error message.
    """
    url = os.getenv("UPLOAD_DOMAIN", "") + "/upload"
    if not url:
        return "ERROR: UPLOAD_DOMAIN environment variable is missing."

    # Sanitize file_name by removing protocol and replacing slashes
    data["file_name"] = data["file_name"].replace("://", "_").replace("/", "_")
    files = {
        'file': (data["file_name"], data["content"], 'text/plain'),
        'file_name': (None, data["file_name"])
    }

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(url, files=files)
            response.raise_for_status()
            result = response.json()
            return f"Uploaded file to the URL: {result['uploaded_file']}"
    except httpx.RequestError as e:
        print("Upload failed:", e)
        return "Upload failed"

async def play_music(data: dict) -> str:
    """
    Asynchronicznie wysyła żądanie do usługi odtwarzania muzyki.

    Parameters:
    - data (dict): JSON payload for the music service.

    Returns:
    - str: The response from the music service or an error message.
    """
    url = os.getenv("MUSIC_URL", "")
    if not url:
        return "ERROR: MUSIC_URL environment variable is missing."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(url, json=data)
            response.raise_for_status()
            result = response.json()
            return result.get("data", "Music playback response received")
    except httpx.RequestError as e:
        print("Error playing music:", e)
        return "Failed to play music"

# Dictionary mapping tool names to functions
tools = {
    'get_html_contents': browse,
    'upload_text_file': upload_file,
    'play_music': play_music
}

1
2
3
4

mkdir ansible_project
cd ansible_project
mkdir roles
mkdir playbooks

1
2
3
4
5

mkdir -p roles/environment/tasks
mkdir -p roles/environment/files
mkdir -p roles/application_files/tasks
mkdir -p roles/application_files/files
mkdir -p roles/server_configuration/tasks

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

---
- name: Utwórz katalog projektu
  file:
    path: "{{ project_dir }}"
    state: directory

- name: Utwórz środowisko wirtualne
  command: python3 -m venv "{{ project_dir }}/venv"
  args:
    creates: "{{ project_dir }}/venv"

- name: Skopiuj plik requirements.txt do katalogu projektu
  copy:
    src: "requirements.txt"  # Ansible automatycznie wyszuka plik w katalogu `files`
    dest: "{{ project_dir }}/requirements.txt"

- name: Aktywuj środowisko i zainstaluj wymagane pakiety
  pip:
    requirements: "{{ project_dir }}/requirements.txt"
    virtualenv: "{{ project_dir }}/venv"
    virtualenv_command: python3 -m venv

 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

---
- name: Skopiuj plik index.py do katalogu projektu
  copy:
    src: roles/application_files/files/index.py
    dest: "{{ project_dir }}/index.py"

- name: Skopiuj plik types.dt.py do katalogu projektu
  copy:
    src: roles/application_files/files/types.dt.py
    dest: "{{ project_dir }}/types.dt.py"

- name: Skopiuj plik README.md do katalogu projektu
  copy:
    src: roles/application_files/files/README.md
    dest: "{{ project_dir }}/README.md"

- name: Skopiuj plik log.md do katalogu projektu
  copy:
    src: roles/application_files/files/log.md
    dest: "{{ project_dir }}/log.md"

- name: Utwórz katalog lib
  file:
    path: "{{ project_dir }}/lib"
    state: directory

- name: Skopiuj plik ai.py do katalogu lib
  copy:
    src: roles/application_files/files/ai.py
    dest: "{{ project_dir }}/lib/ai.py"

- name: Skopiuj plik prompts.py do katalogu lib
  copy:
    src: roles/application_files/files/prompts.py
    dest: "{{ project_dir }}/lib/prompts.py"

- name: Skopiuj plik agent.py do katalogu lib
  copy:
    src: roles/application_files/files/agent.py
    dest: "{{ project_dir }}/lib/agent.py"

- name: Skopiuj plik tools.py do katalogu lib
  copy:
    src: roles/application_files/files/tools.py
    dest: "{{ project_dir }}/lib/tools.py"

- name: Skopiuj plik task_manager.py do katalogu projektu
  copy:
    src: roles/application_files/files/task_manager.py
    dest: "{{ project_dir }}/task_manager.py"

- name: Skopiuj plik ssh_manager.py do katalogu projektu
  copy:
    src: roles/application_files/files/ssh_manager.py
    dest: "{{ project_dir }}/ssh_manager.py"

- name: Skopiuj plik asgi_app.py do katalogu projektu
  copy:
    src: roles/application_files/files/asgi_app.py
    dest: "{{ project_dir }}/asgi_app.py"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

---
- name: Skonfiguruj plik .env
  copy:
    dest: "{{ project_dir }}/.env"
    content: |
      UPLOAD_DOMAIN=http://localhost:3000
      ANTHROPIC_API_KEY=your_anthropic_api_key

- name: Uruchom serwer aplikacji z Uvicorn
  shell: |
    source "{{ project_dir }}/venv/bin/activate"
    nohup uvicorn index:app --host 0.0.0.0 --port 3000 &
  args:
    executable: /bin/bash
  register: app_start_result
  ignore_errors: true  # Ignoruj błędy na wypadek ponownego uruchomienia

1
2
3
4
5
6
7
8
9

---
- name: Konfiguracja i instalacja projektu AI Dev Agent
  hosts: localhost
  vars:
    project_dir: "/home/adrian/aidevs/agent"
  roles:
    - role: environment
    - role: application_files
    - role: server_configuration

1

zip -r ansible_project.zip ansible_project/