Self-Hosted Personal Agents Basics

# Optional local env for later experiments (not required to learn the patterns)
python -m venv .venv && source .venv/bin/activate
pip install pydantic openai

Host: long-running process on my laptop
Wake: Telegram DM from allowlisted user id only
Tools day one: none beyond post_reply
Model: OpenAI-compatible endpoint (direct or gateway)
Stop: final answer or max_turns=6
Success: I get a useful reply in Telegram without opening a web UI

from pydantic import BaseModel
 
class ChatEvent(BaseModel):
    platform: str          # "telegram"
    sender_id: str
    text: str
    thread_id: str | None = None
    message_id: str
 
def from_telegram_update(update: dict) -> ChatEvent:
    msg = update["message"]
    return ChatEvent(
        platform="telegram",
        sender_id=str(msg["from"]["id"]),
        text=msg.get("text") or "",
        thread_id=str(msg["chat"]["id"]),
        message_id=str(msg["message_id"]),
    )

ALLOWED = {"123456789"}  # your Telegram user id
 
def authorize(event: ChatEvent) -> bool:
    return event.sender_id in ALLOWED
 
def handle(event: ChatEvent) -> str | None:
    if not authorize(event):
        return None  # silent ignore or static "unauthorized"
    return run_agent(event.text)

def run_agent(goal: str, tools: dict, max_turns: int = 6) -> str:
    state = {"goal": goal, "trace": []}
    for turn in range(max_turns):
        decision = decide(state, tools)  # model call
        if decision.done:
            return decision.answer
        obs = tools[decision.tool](**decision.args)
        state["trace"].append({"turn": turn, "tool": decision.tool, "obs": obs})
    return "stopped: max turns - partial result"

def post_reply(thread_id: str, text: str) -> dict:
    # Platform Bot API or SDK; return ok + message id
    return {"ok": True, "thread_id": thread_id, "chars": len(text)}
 
# Day-one tool registry:
# tools = {"post_reply": post_reply}  # or host posts after the loop

import json
from pathlib import Path
 
PREFS_PATH = Path.home() / ".personal-agent" / "prefs.json"
 
DEFAULT_PREFS = {
    "timezone": "America/Los_Angeles",
    "tone": "concise, no emojis",
    "quiet_hours": ["22:00", "07:00"],
}
 
def load_prefs() -> dict:
    if not PREFS_PATH.exists():
        PREFS_PATH.parent.mkdir(parents=True, exist_ok=True)
        PREFS_PATH.write_text(json.dumps(DEFAULT_PREFS, indent=2))
    return json.loads(PREFS_PATH.read_text())
 
# Inject a short prefs block into system context each wake.

from pathlib import Path
 
FLAG = Path.home() / ".personal-agent" / "PAUSED"
 
def is_paused() -> bool:
    return FLAG.exists()
 
def handle(event: ChatEvent) -> str | None:
    if event.text.strip().upper() == "/PAUSE":
        FLAG.parent.mkdir(parents=True, exist_ok=True)
        FLAG.write_text("paused\n")
        return "Agent paused. Send /RESUME to continue."
    if event.text.strip().upper() == "/RESUME":
        FLAG.unlink(missing_ok=True)
        return "Agent resumed."
    if is_paused():
        return "Agent is paused. Send /RESUME."
    if not authorize(event):
        return None
    return run_agent(event.text)

import os
from openai import OpenAI
 
client = OpenAI(
    base_url=os.environ.get("MODEL_BASE_URL", "https://openrouter.ai/api/v1"),
    api_key=os.environ["MODEL_API_KEY"],
)
 
def complete(messages: list[dict]) -> str:
    resp = client.chat.completions.create(
        model=os.environ.get("MODEL_ID", "openai/gpt-4o-mini"),  # verify at build
        messages=messages,
    )
    return resp.choices[0].message.content or ""