import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

def ask_claude(message: str, tools: list = None, conversation_history: list = None):
    messages = conversation_history or []
    messages.append({"role": "user", "content": message})
    
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=4096,
        tools=tools or [],
        messages=messages
    )
    return response

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

def ask_gpt(message: str, tools: list = None):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}],
        tools=tools
    )
    return response

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull a capable model
ollama pull llama3.1:70b

tools = [
    {
        "name": "send_email",
        "description": "Send an email to a specified recipient",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "Recipient email"},
                "subject": {"type": "string", "description": "Email subject"},
                "body": {"type": "string", "description": "Email body"}
            },
            "required": ["to", "subject", "body"]
        }
    },
    {
        "name": "read_file",
        "description": "Read the contents of a file on the local system",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path to read"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "run_shell_command",
        "description": "Execute a shell command on the local system",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string", "description": "Shell command to execute"}
            },
            "required": ["command"]
        }
    },
    {
        "name": "web_search",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    }
]

import json
import subprocess

async def agent_loop(user_message: str, conversation_history: list):
    """
    The core agentic loop - keeps running until the AI 
    produces a final text response (no more tool calls).
    """
    conversation_history.append({
        "role": "user", 
        "content": user_message
    })
    
    while True:
        response = ask_claude(
            message=user_message,
            tools=tools,
            conversation_history=conversation_history
        )
        
        # Check if the AI wants to use tools
        tool_use_blocks = [
            block for block in response.content 
            if block.type == "tool_use"
        ]
        
        if not tool_use_blocks:
            # No tools needed - return the text response
            final_text = next(
                (b.text for b in response.content if b.type == "text"), 
                ""
            )
            conversation_history.append({
                "role": "assistant", 
                "content": response.content
            })
            return final_text
        
        # Execute each tool call
        conversation_history.append({
            "role": "assistant", 
            "content": response.content
        })
        
        tool_results = []
        for tool_block in tool_use_blocks:
            result = await execute_tool(
                tool_block.name, 
                tool_block.input
            )
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool_block.id,
                "content": str(result)
            })
        
        conversation_history.append({
            "role": "user", 
            "content": tool_results
        })


async def execute_tool(tool_name: str, tool_input: dict):
    """Execute a tool and return results."""
    
    if tool_name == "send_email":
        return await send_email_handler(tool_input)
    
    elif tool_name == "read_file":
        try:
            with open(tool_input["path"], "r") as f:
                return f.read()
        except Exception as e:
            return f"Error: {e}"
    
    elif tool_name == "run_shell_command":
        result = subprocess.run(
            tool_input["command"], 
            shell=True, 
            capture_output=True, 
            text=True,
            timeout=30
        )
        return result.stdout or result.stderr
    
    elif tool_name == "web_search":
        return await web_search_handler(tool_input["query"])
    
    return "Unknown tool"

import json
from datetime import datetime

class MemoryManager:
    def __init__(self, memory_file="memory.json"):
        self.memory_file = memory_file
        self.memories = self._load()
    
    def _load(self):
        try:
            with open(self.memory_file, "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return {
                "user_profile": {},
                "preferences": {},
                "facts": [],
                "conversation_summaries": []
            }
    
    def save(self):
        with open(self.memory_file, "w") as f:
            json.dump(self.memories, f, indent=2)
    
    def add_fact(self, fact: str, category: str = "general"):
        self.memories["facts"].append({
            "fact": fact,
            "category": category,
            "timestamp": datetime.now().isoformat()
        })
        self.save()
    
    def update_profile(self, key: str, value: str):
        self.memories["user_profile"][key] = value
        self.save()
    
    def get_context_string(self) -> str:
        """Generate a context string to inject into LLM prompts."""
        profile = self.memories["user_profile"]
        facts = self.memories["facts"][-50:]  # Last 50 facts
        
        context = "## What I Know About You:\n"
        for key, value in profile.items():
            context += f"- {key}: {value}\n"
        
        context += "\n## Things I Remember:\n"
        for fact in facts:
            context += f"- {fact['fact']}\n"
        
        return context

import chromadb

class VectorMemory:
    def __init__(self):
        self.client = chromadb.PersistentClient(path="./memory_db")
        self.collection = self.client.get_or_create_collection(
            name="assistant_memory"
        )
    
    def store(self, text: str, metadata: dict = None):
        self.collection.add(
            documents=[text],
            metadatas=[metadata or {}],
            ids=[f"mem_{datetime.now().timestamp()}"]
        )
    
    def recall(self, query: str, n_results: int = 5):
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        return results["documents"][0]

from telegram import Update
from telegram.ext import Application, MessageHandler, filters

async def handle_message(update: Update, context):
    user_message = update.message.text
    user_id = update.effective_user.id
    
    # Load user's conversation history
    history = load_history(user_id)
    
    # Run through agent loop
    response = await agent_loop(user_message, history)
    
    # Save updated history
    save_history(user_id, history)
    
    # Reply on Telegram
    await update.message.reply_text(response)

app = Application.builder().token("YOUR_BOT_TOKEN").build()
app.add_handler(MessageHandler(filters.TEXT, handle_message))
app.run_polling()

from fastapi import FastAPI, Request
from twilio.rest import Client

app = FastAPI()
twilio_client = Client("ACCOUNT_SID", "AUTH_TOKEN")

@app.post("/webhook/whatsapp")
async def whatsapp_webhook(request: Request):
    form = await request.form()
    message = form.get("Body")
    sender = form.get("From")
    
    # Process through agent
    response = await agent_loop(message, load_history(sender))
    
    # Reply via WhatsApp
    twilio_client.messages.create(
        body=response,
        from_="whatsapp:+14155238886",
        to=sender
    )
    return {"status": "ok"}

import discord

intents = discord.Intents.default()
intents.message_content = True
client = discord.Client(intents=intents)

@client.event
async def on_message(message):
    if message.author == client.user:
        return
    
    response = await agent_loop(
        message.content, 
        load_history(message.author.id)
    )
    await message.channel.send(response)

client.run("YOUR_DISCORD_TOKEN")

import importlib
import os

class SkillManager:
    def __init__(self, skills_dir="skills"):
        self.skills_dir = skills_dir
        self.skills = {}
        self.load_all_skills()
    
    def load_all_skills(self):
        """Dynamically load all skills from the skills directory."""
        for filename in os.listdir(self.skills_dir):
            if filename.endswith(".py") and not filename.startswith("_"):
                skill_name = filename[:-3]
                self.load_skill(skill_name)
    
    def load_skill(self, skill_name: str):
        spec = importlib.util.spec_from_file_location(
            skill_name, 
            f"{self.skills_dir}/{skill_name}.py"
        )
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        self.skills[skill_name] = {
            "name": module.SKILL_NAME,
            "description": module.SKILL_DESCRIPTION,
            "tools": module.TOOLS,
            "execute": module.execute
        }
    
    def get_all_tools(self) -> list:
        """Get tool definitions from all loaded skills."""
        all_tools = []
        for skill in self.skills.values():
            all_tools.extend(skill["tools"])
        return all_tools

# skills/gmail_manager.py

SKILL_NAME = "Gmail Manager"
SKILL_DESCRIPTION = "Read, send, and organize Gmail messages"

TOOLS = [
    {
        "name": "gmail_read_inbox",
        "description": "Read recent emails from Gmail inbox",
        "input_schema": {
            "type": "object",
            "properties": {
                "count": {"type": "integer", "default": 10}
            }
        }
    },
    {
        "name": "gmail_send",
        "description": "Send an email via Gmail",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"}
            },
            "required": ["to", "subject", "body"]
        }
    }
]

async def execute(tool_name: str, params: dict):
    if tool_name == "gmail_read_inbox":
        # Gmail API integration
        return await read_inbox(params.get("count", 10))
    elif tool_name == "gmail_send":
        return await send_email(params)

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

# Morning briefing at 8 AM
@scheduler.scheduled_job('cron', hour=8, minute=0)
async def morning_briefing():
    """Proactively send morning briefing."""
    briefing = await agent_loop(
        "Generate my morning briefing: weather, calendar, "
        "important emails, and tasks for today.",
        system_history
    )
    await send_to_user(briefing)

# Check emails every 30 minutes  
@scheduler.scheduled_job('interval', minutes=30)
async def email_monitor():
    """Monitor inbox for important emails."""
    result = await agent_loop(
        "Check my inbox for any urgent or important emails "
        "that arrived in the last 30 minutes. Only notify me "
        "if something truly needs my attention.",
        system_history
    )
    if "urgent" in result.lower() or "important" in result.lower():
        await send_to_user(result)

# Heartbeat check-in
@scheduler.scheduled_job('cron', hour=14, minute=0)
async def heartbeat():
    """Afternoon check-in."""
    await send_to_user(
        "Hey! Quick afternoon check-in 🦞 "
        "Anything I can help you with?"
    )

scheduler.start()

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load skills, start scheduler
    skill_manager.load_all_skills()
    scheduler.start()
    yield
    # Shutdown
    scheduler.shutdown()

app = FastAPI(lifespan=lifespan)

@app.post("/webhook/telegram")
async def telegram_webhook(request: Request):
    # Handle Telegram messages
    pass

@app.post("/webhook/whatsapp")
async def whatsapp_webhook(request: Request):
    # Handle WhatsApp messages
    pass

@app.get("/health")
async def health():
    return {"status": "alive", "skills_loaded": len(skill_manager.skills)}

{
  "@context": "https://schema.org",
  "@type": "HowTo",
  "name": "How to Build Your Own AI Personal Assistant Like OpenClaw",
  "description": "Step-by-step guide to building a personal AI assistant similar to OpenClaw using Claude AI, Python, and FastAPI.",
  "step": [
    {"@type": "HowToStep", "name": "Choose Your AI Brain", "text": "Select an LLM backend: Claude AI, OpenAI, or local models"},
    {"@type": "HowToStep", "name": "Build the Agent Core", "text": "Create the agentic loop with tool calling capabilities"},
    {"@type": "HowToStep", "name": "Add Persistent Memory", "text": "Implement file-based or vector memory system"},
    {"@type": "HowToStep", "name": "Connect Messaging Channels", "text": "Integrate with WhatsApp, Telegram, or Discord"},
    {"@type": "HowToStep", "name": "Create Skill System", "text": "Build extensible plugin architecture"},
    {"@type": "HowToStep", "name": "Add Proactive Capabilities", "text": "Implement scheduled tasks and heartbeats"},
    {"@type": "HowToStep", "name": "Deploy and Secure", "text": "Deploy to production with security best practices"}
  ],
  "author": {"@type": "Organization", "name": "AI By Tech", "url": "https://aibytec.com"},
  "datePublished": "2026-02-07"
}