build-your-own-openclaw-agent-tutorial

# Build Your Own OpenClaw Agent Tutorial > Skill by [ara.so](https://ara.so) — Hermes Skills collection. A comprehensive tutorial for building AI agents progressively, from a basic chat loop to a production-ready autonomous agent system. This project walks through 18 steps covering single-agent capabilities, event-driven architecture, multi-agent collaboration, and production features like memory and concurrency control. ## What This Tutorial Teaches The tutorial is organized into 4 phases: 1. **Phase 1 (Steps 0-6)**: Single agent with tools, skills, persistence, and web access 2. **Phase 2 (Steps 7-10)**: Event-driven architecture with multi-platform support 3. **Phase 3 (Steps 11-15)**: Autonomous agents with routing and collaboration 4. **Phase 4 (Steps 16-17)**: Production features like concurrency and long-term memory ## Initial Setup ### Clone the Repository ```bash git clone https://github.com/czl9707/build-your-own-openclaw.git cd build-your-own-openclaw ``` ### Configure API Keys ```bash # Copy example config cp default_workspace/config.example.yaml default_workspace/config.user.yaml ``` Edit `default_workspace/config.user.yaml`: ```yaml llm: model: "gpt-4" # or anthropic/claude-3-5-sonnet-20241022, etc. api_key: "${OPENAI_API_KEY}" # Use environment variable # See https://docs.litellm.ai/docs/providers for all providers # Optional: Add additional services web: search_api_key: "${SERPER_API_KEY}" ``` ### Install Dependencies (for each step) ```bash cd 00-chat-loop # or any step directory pip install -r requirements.txt ``` ## Phase 1: Building a Capable Single Agent ### Step 0: Basic Chat Loop The foundation - a simple conversation loop with an LLM. ```python # 00-chat-loop/main.py from litellm import completion def chat_loop(): messages = [] while True: user_input = input("You: ") if user_input.lower() in ['/exit', '/quit']: break messages.append({"role": "user", "content": user_input}) response = completion( model="gpt-4", messages=messages, api_key="${OPENAI_API_KEY}" ) assistant_message = response.choices[0].message.content messages.append({"role": "assistant", "content": assistant_message}) print(f"Assistant: {assistant_message}") if __name__ == "__main__": chat_loop() ``` Run it: ```bash cd 00-chat-loop python main.py ``` ### Step 1: Adding Tools Give your agent function-calling capabilities. ```python # 01-tools/tools.py def get_current_weather(location: str) -> dict: """Get the current weather for a location.""" # Tool implementation return {"location": location, "temperature": 72, "condition": "sunny"} # Tool schema for LLM weather_tool = { "type": "function", "function": { "name": "get_current_weather", "description": "Get the current weather in a given location", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "City name, e.g. San Francisco" } }, "required": ["location"] } } } ``` Using tools in the chat loop: ```python import json from litellm import completion response = completion( model="gpt-4", messages=messages, tools=[weather_tool], tool_choice="auto" ) # Handle tool calls if response.choices[0].message.tool_calls: for tool_call in response.choices[0].message.tool_calls: function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) # Execute tool result = get_current_weather(**arguments) # Add tool result to messages messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result) }) ``` ### Step 2: Skills with SKILL.md Extend agent capabilities through markdown skill files. ```markdown <!-- skills/web_search.md --> # Web Search Skill You can search the internet using the `search_web` tool. ## When to Use - User asks for current information - Need to verify facts - Looking for recent news ## Example User: "What's the latest news on AI?" You: Let me search for that. [calls search_web("latest AI news")] ``` Loading skills: ```python # 02-skills/skill_loader.py import os def load_skills(skills_dir="skills"): """Load all .md files from skills directory.""" skills_content = [] for filename in os.listdir(skills_dir): if filename.endswith(".md"): with open(os.path.join(skills_dir, filename), 'r') as f: skills_content.append(f.read()) return "\n\n".join(skills_content) # Add to system prompt system_prompt = f"""You are a helpful assistant. ## Your Skills {load_skills()} """ ``` ### Step 3: Conversation Persistence Save conversations to resume later. ```python # 03-persistence/session_manager.py import json from datetime import datetime from pathlib import Path class SessionManager: def __init__(self, sessions_dir="sessions"): self.sessions_dir = Path(sessions_dir) self.sessions_dir.mkdir(exist_ok=True) def save_session(self, session_id: str, messages: list): """Save conversation history.""" session_file = self.sessions_dir / f"{session_id}.json" data = { "session_id": session_id, "updated_at": datetime.now().isoformat(), "messages": messages } with open(session_file, 'w') as f: json.dump(data, f, indent=2) def load_session(self, session_id: str) -> list: """Load conversation history.""" session_file = self.sessions_dir / f"{session_id}.json" if not session_file.exists(): return [] with open(session_file, 'r') as f: data = json.load(f) return data.get("messages", []) def list_sessions(self) -> list: """List all available sessions.""" return [f.stem for f in self.sessions_dir.glob("*.json")] ``` Usage: ```python manager = SessionManager() # Load or create session session_id = "my-conversation" messages = manager.load_session(session_id) # After each exchange manager.save_session(session_id, messages) ``` ### Step 4: Slash Commands Direct user control over agent behavior. ```python # 04-slash-commands/commands.py class CommandHandler: def __init__(self, session_manager): self.session_manager = session_manager self.commands = { '/new': self.new_session, '/load': self.load_session, '/list': self.list_sessions, '/save': self.save_session, '/clear': self.clear_session, '/help': self.show_help } def handle(self, user_input: str, current_session: str, messages: list): """Handle slash commands.""" parts = user_input.split() command = parts[0] args = parts[1:] if len(parts) > 1 else [] if command in self.commands: return self.commands[command](args, current_session, messages) return None # Not a command def new_session(self, args, current_session, messages): new_id = args[0] if args else f"session_{int(time.time())}" return {"action": "new_session", "session_id": new_id} def load_session(self, args, current_session, messages): if not args: print("Usage: /load <session_id>") return {"action": "none"} loaded = self.session_manager.load_session(args[0]) return {"action": "load_session", "session_id": args[0], "messages": loaded} ``` ### Step 5: Context Compaction Manage token limits by summarizing old messages. ```python # 05-compaction/compactor.py from litellm import completion class MessageCompactor: def __init__(self, max_messages=20): self.max_messages = max_messages def compact_if_needed(self, messages: list) -> list: """Compact messages if they exceed threshold.""" if len(messages) <= self.max_messages: return messages # Keep system message and recent messages system_msgs = [m for m in messages if m["role"] == "system"] recent_msgs = messages[-(self.max_messages - 2):] # Summarize older messages old_msgs = messages[len(system_msgs):-len(recent_msgs)] summary = self._summarize_messages(old_msgs) return system_msgs + [ {"role": "system", "content": f"Previous conversation summary:\n{summary}"} ] + recent_msgs def _summarize_messages(self, messages: list) -> str: """Generate summary of message history.""" conversation = "\n".join([ f"{m['role']}: {m['content']}" for m in messages ]) response = completion( model="gpt-4", messages=[{ "role": "user", "content": f"Summarize this conversation concisely:\n\n{conversation}" }] ) return response.choices[0].message.content ``` ### Step 6: Web Tools Give your agent internet access. ```python # 06-web-tools/web_tools.py import requests import os def search_web(query: str, num_results: int = 5) -> list: """Search the web using Serper API.""" api_key = os.getenv("SERPER_API_KEY") response = requests.post( "https://google.serper.dev/search", headers={"X-API-KEY": api_key}, json={"q": query, "num": num_results} ) results = response.json() return [ { "title": r.get("title"), "link": r.get("link"), "snippet": r.get("snippet") } for r in results.get("organic", []) ] def fetch_webpage(url: str) -> str: """Fetch and extract text from a webpage.""" from bs4 import BeautifulSoup response = requests.get(url, timeout=10) soup = BeautifulSoup(response.content, 'html.parser') # Remove script and style elements for script in soup(["script", "style"]): script.decompose() return soup.get_text(separator="\n", strip=True) ``` ## Phase 2: Event-Driven Architecture ### Step 7: Event-Driven Refactor Decouple components with an event bus. ```python # 07-event-driven/event_bus.py from typing import Callable, Dict, List from dataclasses import dataclass from enum import Enum class EventType(Enum): MESSAGE_RECEIVED = "message_received" MESSAGE_SENT = "message_sent" TOOL_CALLED = "tool_called" SESSION_CREATED = "session_created" @dataclass class Event: type: EventType data: dict source: str class EventBus: def __init__(self): self.listeners: Dict[EventType, List[Callable]] = {} def subscribe(self, event_type: EventType, handler: Callable): """Subscribe to an event type.""" if event_type not in self.listeners: self.listeners[event_type] = [] self.listeners[event_type].append(handler) def publish(self, event: Event): """Publish an event to all subscribers.""" if event.type in self.listeners: for handler in self.listeners[event.type]: handler(event) # Usage