Building Self Hosted AI Agents with Python

· 9 min read ·
Building Self Hosted AI Agents with Python

What You’ll Need

To follow this tutorial, you will need the following infrastructure and services:

  • Hetzner VPS or Contabo VPS for hosting your agent execution engine (8GB+ RAM recommended for local model hosting or 2GB for API proxying)
  • DigitalOcean as an alternative cloud provider
  • n8n Cloud or self-hosted n8n for triggering webhook events
  • Namecheap if you plan to attach a custom SSL domain to your backend API engine
  • Python 3.10+ installed on your local workstation or server

Table of Contents

Architecture of a Self-Hosted AI Agent

Relying on commercial agent frameworks often locks you into subscription tiers, rate limits, and black-box orchestration logic. When I build production-grade autonomous systems, I prefer complete control over the execution loop, vector stores, and tool execution environments.

Building your own agent framework allows you to inspect every prompt sent to the LLM, strictly sanitize dynamic tool executions, and eliminate recurring software overhead. If you are analyzing cloud infrastructure costs, reading our guide on How to Reduce Your SaaS Bill with Self-Hosted Alternatives explains why running workloads on bare-metal or unmanaged VPS hardware offers massive savings over closed SaaS wrappers.

A robust self-hosted AI agent consists of four fundamental components:

  1. State and Context Memory Manager: Captures session histories and persists task context using SQLite or PostgreSQL.
  2. LLM Orchestration Layer: Sends strict JSON schemas to the model and parses structured tool invocation outputs.
  3. Execution Sandbox: Runs Python code, local database queries, or network requests based on model decisions.
  4. Trigger Interface: Exposes webhooks or scheduled jobs to initiate autonomous task runs.
+-----------------------------------------------------------------+
|                         Trigger Event                           |
|                 (REST Webhook / Scheduled Cron)                 |
+-----------------------------------------------------------------+
                                |
                                v
+-----------------------------------------------------------------+
|                    Python Execution Engine                      |
|  +-------------------+   +------------------+   +------------+  |
|  | State Repository  |   | Function Calling |   | Dynamic    |  |
|  |  (SQLite Store)   |   |   JSON Schema    |   | Tools Engine|  |
|  +-------------------+   +------------------+   +------------+  |
+-----------------------------------------------------------------+
                                |
                                v
+-----------------------------------------------------------------+
|                   LLM Engine (Ollama / vLLM)                    |
+-----------------------------------------------------------------+

By decoupling these layers, you can swap local inference servers like Ollama or vLLM with cloud endpoints without rewriting your underlying tool integration code.

Provisioning the Cloud Infrastructure

To run both an inference engine and the agent orchestration runtime without bottlenecks, you need dedicated computing power. I recommend launching a virtual instance on a Hetzner VPS or a high-RAM instance on Contabo VPS . If you already operate within an existing cloud ecosystem, a compute droplet on DigitalOcean works just as well.

Once your Linux server is online, SSH into the machine and install Python dependencies along with Ollama for hosting local open-source weights (such as Llama 3 or Qwen 2.5):

sudo apt update && sudo apt upgrade -y
sudo apt install python3-pip python3-venv sqlite3 build-essential -y

curl -fsSL https://ollama.com/install.sh | sh

ollama pull qwen2.5:7b-instruct

Verify that the Ollama service is listening locally on port 11434 before proceeding:

curl http://localhost:11434/api/tags

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Writing the Core Python Agent Runtime

We will build a custom Python framework that implements an agentic loop: Reasoning -> Tool Selection -> Tool Execution -> Final Response.

Create a dedicated workspace directory and setup a virtual environment:

mkdir self_hosted_agent
cd self_hosted_agent
python3 -m venv venv
source venv/bin/activate
pip install requests pydantic httpx

Now create the file agent_runtime.py. This script implements state storage with SQLite, defines available function interfaces for the LLM, manages the conversational state loop, and executes tools dynamic without relying on external frameworks like LangChain or AutoGen.

import json
import sqlite3
import httpx
from typing import List, Dict, Any, Callable

DB_FILE = "agent_state.db"

def init_db():
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS system_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            action TEXT NOT NULL,
            details TEXT NOT NULL
        )
    """)
    conn.commit()
    conn.close()

def log_system_action(action: str, details: str) -> str:
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO system_logs (action, details) VALUES (?, ?)",
        (action, details)
    )
    conn.commit()
    conn.close()
    return f"Successfully logged action: {action}"

def query_system_logs(limit: int = 5) -> str:
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT timestamp, action, details FROM system_logs ORDER BY id DESC LIMIT ?",
        (limit,)
    )
    rows = cursor.fetchall()
    conn.close()
    return json.dumps([{"timestamp": r[0], "action": r[1], "details": r[2]} for r in rows])

def execute_http_get(url: str) -> str:
    try:
        response = httpx.get(url, timeout=10.0)
        return json.dumps({"status_code": response.status_code, "body": response.text[:500]})
    except Exception as e:
        return json.dumps({"error": str(e)})

AVAILABLE_TOOLS: Dict[str, Callable] = {
    "log_system_action": log_system_action,
    "query_system_logs": query_system_logs,
    "execute_http_get": execute_http_get
}

TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "log_system_action",
            "description": "Log an operational event or action into the local database.",
            "parameters": {
                "type": "object",
                "properties": {
                    "action": {"type": "string", "description": "Short category name for the action"},
                    "details": {"type": "string", "description": "Detailed description of the logged event"}
                },
                "required": ["action", "details"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_system_logs",
            "description": "Retrieve recent log records from the persistent SQLite database.",
            "parameters": {
                "type": "object",
                "properties": {
                    "limit": {"type": "integer", "description": "Number of logs to fetch"}
                },
                "required": ["limit"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "execute_http_get",
            "description": "Perform an HTTP GET request to fetch external web content or API status.",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "The fully qualified URL to reach"}
                },
                "required": ["url"]
            }
        }
    }
]

class SelfHostedAgent:
    def __init__(self, model_name: str = "qwen2.5:7b-instruct", api_url: str = "http://localhost:11434/api/chat"):
        self.model_name = model_name
        self.api_url = api_url
        self.messages: List[Dict[str, Any]] = [
            {"role": "system", "content": "You are a self-hosted autonomous agent. You complete tasks by invoking system tools. Always use available functions when appropriate."}
        ]
        init_db()

    def run(self, user_prompt: str, max_iterations: int = 5):
        self.messages.append({"role": "user", "content": user_prompt})
        
        for iteration in range(max_iterations):
            payload = {
                "model": self.model_name,
                "messages": self.messages,
                "tools": TOOL_SCHEMAS,
                "stream": False
            }
            
            response = httpx.post(self.api_url, json=payload, timeout=60.0)
            if response.status_code != 200:
                raise RuntimeError(f"Ollama API request failed: {response.text}")

            result = response.json()
            message = result.get("message", {})
            self.messages.append(message)

            tool_calls = message.get("tool_calls")
            if not tool_calls:
                return message.get("content")

            for tool_call in tool_calls:
                func_name = tool_call["function"]["name"]
                raw_args = tool_call["function"]["arguments"]
                
                if isinstance(raw_args, str):
                    args = json.loads(raw_args)
                else:
                    args = raw_args

                if func_name in AVAILABLE_TOOLS:
                    execution_result = AVAILABLE_TOOLS[func_name](**args)
                else:
                    execution_result = f"Error: Tool '{func_name}' is not registered in runtime."

                self.messages.append({
                    "role": "tool",
                    "content": str(execution_result)
                })

        return "Agent execution terminated: maximum reasoning iterations reached."

if __name__ == "__main__":
    agent = SelfHostedAgent()
    print("--- Executing Initial Task ---")
    prompt = "Log an event named 'BACKUP_COMPLETE' with details 'Database backup stored successfully', then query the last log."
    final_output = agent.run(prompt)
    print("\nAgent Response:\n" + str(final_output))

Implementing Dynamic Tool Execution

The core script defined above gives the model direct control over structured code execution. When the LLM decides it needs real-time information or database modifications, it emits a tool_calls payload instead of simple plain text.

Let’s break down the mechanics of the execution loop inside SelfHostedAgent:

  1. State Injection: The full message history (including user instructions, previous assistant calls, and tool returns) is passed on every turn.
  2. Schema Matching: Native tools strictly match OpenAPI-style parameters so local models accurately predict JSON shapes.
  3. Execution Injection: When the model returns a call for log_system_action, our runtime routes parameters into standard Python functions, captures output strings, and feeds them back into context with the "role": "tool" designation.

To test this execution loop, launch the script directly:

python3 agent_runtime.py

The terminal output will display the full cycle of the model calling the SQLite write tool, receiving confirmation, executing the read tool, and returning a final summary back to the user interface.

Model Selection: Local vs Hosted LLMs

Self-hosting gives you the freedom to choose your LLM engine based on latency, memory availability, and task complexity.

While hosting models locally on a GPU-backed server offers privacy and eliminates API fees, smaller parameters (like 7B or 8B weights) occasionally hallucinate complex nested tool schemas. When building production workflows where deterministic function parsing is required, hybrid setups can prove useful.

+-------------------+--------------------+------------------------+--------------------+
| Model Choice      | Hosting Strategy   | Function Calling Skill | Latency Profile    |
+-------------------+--------------------+------------------------+--------------------+
| Qwen 2.5 7B       | Fully Local VPS    | Moderate to High       | 300ms - 1.2s       |
| Llama 3.1 8B      | Fully Local VPS    | High                   | 400ms - 1.5s       |
| GPT-4o Mini       | Cloud API Wrapper  | Very High              | 200ms - 600ms      |
| Claude 3.5 Haiku  | Cloud API Wrapper  | Elite                  | 150ms - 400ms      |
+-------------------+--------------------+------------------------+--------------------+

If you decide to delegate standard reasoning tasks to low-cost cloud APIs while keeping data ingestion self-hosted, compare cost profiles by reading our analysis on Claude Haiku vs GPT-4o Mini for Automation Pipelines .

Deploying and Automating Agent Workflows

To make your Python agent operational around the clock, expose its control loop via an HTTP API microservice using FastAPI and schedule routine operations.

First, update your virtual environment with web server dependencies:

pip install fastapi uvicorn

Create server.py to wrap your agent inside a web service:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent_runtime import SelfHostedAgent

app = FastAPI(title="Self-Hosted AI Agent API")
agent = SelfHostedAgent()

class TaskRequest(BaseModel):
    prompt: str

class TaskResponse(BaseModel):
    result: str

@app.post("/run-agent", response_model=TaskResponse)
def handle_task(request: TaskRequest):
    try:
        output = agent.run(request.prompt)
        return TaskResponse(result=output)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

To run this backend continuously on Linux without manual CLI intervention, create a standard systemd service file at /etc/systemd/system/ai-agent.service:

[Unit]
Description=Self Hosted Python AI Agent Engine
After=network.target

[Service]
User=root
WorkingDirectory=/root/self_hosted_agent
ExecStart=/root/self_hosted_agent/venv/bin/uvicorn server:app --host 0.0.0.0 --port 8000
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Reload systemd daemon, enable, and start your backend service:

sudo systemctl daemon-reload
sudo systemctl enable ai-agent
sudo systemctl start ai-agent

With your API running on port 8000, you can trigger autonomous tasks using simple HTTP POST requests. If you want to connect this HTTP service to cron schedules, webhooks, or visual logic flows, deploy an instance of n8n Cloud or host n8n on your server.

Alternatively, if you prefer handling system execution purely inside Python scripts without web servers, review our guide on Scheduling Python Scripts for Automated Tasks to configure cron drivers and system daemons directly.

Getting Started

Building a self-hosted AI agent provides full ownership over data flow, tool integrations, and operational costs. By bypassing third-party platform limitations, you build robust infrastructure tailored specifically to your data schema and processing requirements.

To start deploying your own agents:

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr — mention SYS3-HUGO for priority. Or DM at chasebot.online .

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online