Build a Personal AI Agent in Python
Build an AI agent with a gateway, tool registry, and Telegram integration.
Introduction
30 Second Summary
Imagine having your own AI assistant that lives on your computer, remembers your preferences, and can actually do things for you. Not just chat, but run commands, manage files, and fetch information from the web.
In this project, you will build a personal AI agent in Python that connects to Telegram and your terminal. It uses the same gateway and tool architecture found in production agents like OpenClaw and Hermes Agent.
What You'll Build
A fully functional AI agent that runs locally, routes messages through a gateway from multiple channels, and executes real actions on your machine through a self-registering tool system.
By the end of this project, you'll have:
- A Gateway that routes messages between a CLI channel, a Telegram channel, and your agent.
- An Agent loop powered by OpenAI function calling that autonomously executes shell commands, reads and writes files, and fetches web pages.
- A Tool Registry with self-registering tools, mirroring the exact pattern used by Hermes Agent's 61-tool system.
- File-based persistent memory so your agent remembers you across sessions.
- Secret Mission: Add a cron scheduler that lets your agent send you proactive messages on Telegram at scheduled intervals.
Are there any prerequisites?
This project assumes intermediate Python knowledge, familiarity with async/await, and a basic understanding of APIs. You will also need a Telegram account and an OpenAI API key.
Before We Start
Before We Start...
Set Up the Project and Dependencies
Every production AI agent has a clear directory structure that separates concerns. Your agent will follow the same layout used by Hermes Agent: a root folder with dedicated directories for channels, tools, and configuration files.
In this step, you will scaffold the project folder, write the agent's personality and memory files, install the Python packages it depends on, and confirm your OpenAI API key works.
In this step, get ready to:
- Create the my-agent/ project directory with channels/, tools/, and cron/ subdirectories.
- Write the configuration files: SOUL.md, MEMORY.md, .env, and requirements.txt.
- Install dependencies and verify your OpenAI API key.
Create the project directory structure
The agent's folder structure mirrors how Hermes Agent separates its code. Each directory holds a single responsibility: channels for messaging adapters, tools for the tool registry, and cron for scheduled tasks.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and its subdirectories by running these commands:
mkdir my-agent
cd my-agent
mkdir channels tools cron- Confirm the directories exist by running:
lsYou should see channels, cron, and tools listed in the output.
Don't see all three folders?
Make sure you ran cd my-agent before running ls. If you're still in ~/Desktop, you will see your Desktop files instead of the project folders.
help me troubleshoot my project directory structure
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Cursor and press Enter to open it.
- In Cursor, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop and select the my-agent folder, then click Open.
You should see the channels, cron, and tools folders in the Explorer sidebar on the left.
Add the persona and dependency files
Your agent needs two context files that shape its behavior. SOUL.md defines the agent's personality and capabilities. MEMORY.md stores persistent facts about you that survive across sessions.
- In Cursor's Explorer sidebar, right-click in the empty space below your folders and select New File.
- Name the file SOUL.md and press Enter.
- Paste the following content into SOUL.md:
You are a helpful personal AI assistant running on the user's local machine.
You can execute shell commands, read and write files, fetch web pages, and remember things.
Be concise and action-oriented. When the user asks you to do something, use your tools.
When you learn something important about the user, use the remember tool to save it.- Create another new file in the same location called MEMORY.md.
- Paste the following content into MEMORY.md:
# Memory
This file stores persistent memories about the user across sessions.What do these files do?
SOUL.md is the agent's system prompt. It tells the LLM what role it plays and what tools it has access to. This is the same pattern Hermes Agent uses to define its persona.
MEMORY.md starts nearly empty. As you chat with your agent, it will append facts here. Because this is a plain file on disk, those memories persist even after you restart the agent.
- Create one more new file called requirements.txt in the root of my-agent.
- Paste the following dependency list into requirements.txt:
openai==2.36.0
python-telegram-bot==22.7
python-dotenv==1.1.0What are these dependencies?
- openai is the official Python SDK for calling OpenAI's Chat Completions API, including function calling.
- python-telegram-bot provides an async framework for receiving and sending Telegram messages.
- python-dotenv loads environment variables from your .env file so you don't have to export them manually in your shell.
- Save all three files by pressing Cmd+S (macOS) or Ctrl+S (Windows) in each file tab.
✔️ Awesome, I've got everything!
Great. Make sure all files are saved before moving on.
ⓧ I'd like to double check the full code
Here is what each file should contain at this point.
You are a helpful personal AI assistant running on the user's local machine.
You can execute shell commands, read and write files, fetch web pages, and remember things.
Be concise and action-oriented. When the user asks you to do something, use your tools.
When you learn something important about the user, use the remember tool to save it.# Memory
This file stores persistent memories about the user across sessions.openai==2.36.0
python-telegram-bot==22.7
python-dotenv==1.1.0Install dependencies and verify your API key
Before your agent can call OpenAI or connect to Telegram, you need to install the packages listed in requirements.txt. You also need a .env file to store your API credentials locally.
- Create a new file called .env in the root of your my-agent folder.
- Paste the following into .env:
OPENAI_API_KEY=your-api-key-here
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-here- Replace your-api-key-here with your real OpenAI API key.
- Leave your-telegram-bot-token-here as-is for now. You will get your Telegram token in a later step.
- Save the file.
Keep your API key private
Never commit your .env file to Git or share it publicly. Your OpenAI key gives full access to your account and will be billed for any usage.
- Open a terminal inside Cursor by pressing Ctrl+` (backtick key, top-left of your keyboard).
- Install the project dependencies by running:
pip install -r requirements.txtYou should see output showing each package downloading and installing. The final line will say Successfully installed followed by the package names.
Seeing 'pip: command not found'?
Try running pip3 install -r requirements.txt instead. On some systems, Python 3's package manager is named pip3.
If you see permissions errors, add --user to the end of the command to install in your user directory.
help me fix my pip install issue
Now confirm your OpenAI API key is valid by making a quick test call.
- Verify your key works by running this test command:
python -c "
from dotenv import load_dotenv
load_dotenv()
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': 'Say hello in one word'}]
)
print(response.choices[0].message.content)
"You should see a short greeting printed in the terminal (something like Hello or Hi). This confirms your API key is valid and the openai package is installed correctly.
What does this test do?
This script loads your .env file, creates an OpenAI client, and sends a one-word prompt to GPT-4o-mini. If the response prints, your key and network connection are both working.
Seeing an authentication error?
Double-check that you replaced your-api-key-here in .env with your actual OpenAI API key. The key starts with sk-.
Make sure there are no extra spaces or quotes around the key in your .env file. The format should be OPENAI_API_KEY=sk-... with no spaces around the = sign.
help me debug my OpenAI API key
✔️ Awesome, I've got everything!
Great. Double check you have saved all your files before moving on.
ⓧ I'd like to double check the full code
Here is what each file should contain at this point.
You are a helpful personal AI assistant running on the user's local machine.
You can execute shell commands, read and write files, fetch web pages, and remember things.
Be concise and action-oriented. When the user asks you to do something, use your tools.
When you learn something important about the user, use the remember tool to save it.# Memory
This file stores persistent memories about the user across sessions.OPENAI_API_KEY=your-api-key-here
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-hereopenai==2.36.0
python-telegram-bot==22.7
python-dotenv==1.1.0Your project is scaffolded and your API connection is confirmed. Next up, you will build the tool registry that gives your agent the power to actually do things on your machine.
Build the Tool Registry and Tools
Your project structure and dependencies are ready. Now it is time to give your agent the ability to actually do things.
In production agents like Hermes Agent, tools are the functions an agent can call to interact with the real world. The tool registry pattern lets each tool self-register simply by being imported. This means adding a new capability to your agent is as simple as creating a new file.
In this step, get ready to:
- Build a tool registry that stores tool definitions and dispatches calls by name.
- Create five self-registering tools: shell commands, file read, file write, web fetch, and memory.
Create the tool registry
The registry is the central hub where all tools live. Each tool provides its name, a description, a JSON Schema defining its parameters, and an async handler function. When OpenAI tells the agent to call a tool, the registry looks it up and runs it.
- Create the empty __init__.py file that makes tools/ a Python package by running this command:
touch tools/__init__.py- Create a new file called tools/registry.py inside your tools/ folder.
- Add the imports and the Tool class by pasting the following code:
import json
from typing import Any, Callable, Awaitable
# Represents a single tool the agent can call
class Tool:
def __init__(
self,
name: str,
description: str,
parameters: dict,
handler: Callable[..., Awaitable[str]],
):
self.name = name
self.description = description
self.parameters = parameters
self.handler = handler
def schema(self) -> dict:
# Returns the OpenAI function-calling format for this tool
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}What does this code do?
The Tool class wraps everything OpenAI needs to know about a tool: its name, what it does, and the shape of its inputs.
The schema() method outputs the exact JSON format that OpenAI's function calling expects. The handler is the async function that actually executes when the tool is called.
You will not see a change yet. The next chunk uses this class.
- Now add the ToolRegistry class and the global registry instance below your Tool class in the same file:
# Central registry that all tools register themselves with
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(
self,
name: str,
description: str,
parameters: dict,
handler: Callable[..., Awaitable[str]],
):
# Store the tool so it can be dispatched later
self._tools[name] = Tool(name, description, parameters, handler)
def get_schemas(self) -> list[dict]:
# Returns all tool schemas for the OpenAI API call
return [tool.schema() for tool in self._tools.values()]
async def dispatch(self, name: str, arguments: str) -> str:
# Find and execute a tool by name, parsing the JSON arguments
if name not in self._tools:
return json.dumps({"error": f"Unknown tool: {name}"})
try:
args = json.loads(arguments)
result = await self._tools[name].handler(**args)
return result
except Exception as e:
return json.dumps({"error": str(e)})
# Global registry instance
registry = ToolRegistry()What does this code do?
- register() stores a new tool in the internal dictionary, keyed by name.
- get_schemas() returns all registered tool schemas in the format OpenAI expects.
- dispatch() finds a tool by name, parses the JSON arguments string, and calls the handler.
- The global registry instance at the bottom is what every tool file imports to register itself.
- Save tools/registry.py.
- Verify the registry loads by running this command:
python -c "from tools.registry import registry; print(registry.get_schemas())"You should see [] printed. That is an empty list because no tools have registered yet.
Seeing a ModuleNotFoundError?
- Make sure you created tools/__init__.py. Without it, Python does not recognize tools/ as a package.
- Confirm you are running the command from the my-agent/ directory, not from inside tools/.
Why can't Python find my tools module?
✔️ Awesome, I've got everything!
Great. Make sure you have saved the file before moving on.
ⓧ I'd like to double check the full code
import json
from typing import Any, Callable, Awaitable
class Tool:
def __init__(
self,
name: str,
description: str,
parameters: dict,
handler: Callable[..., Awaitable[str]],
):
self.name = name
self.description = description
self.parameters = parameters
self.handler = handler
def schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(
self,
name: str,
description: str,
parameters: dict,
handler: Callable[..., Awaitable[str]],
):
self._tools[name] = Tool(name, description, parameters, handler)
def get_schemas(self) -> list[dict]:
return [tool.schema() for tool in self._tools.values()]
async def dispatch(self, name: str, arguments: str) -> str:
if name not in self._tools:
return json.dumps({"error": f"Unknown tool: {name}"})
try:
args = json.loads(arguments)
result = await self._tools[name].handler(**args)
return result
except Exception as e:
return json.dumps({"error": str(e)})
# Global registry instance
registry = ToolRegistry()Create the shell tool
Your first tool gives the agent the power to run shell commands on your machine. This is what lets the agent list files, check system info, or run scripts when you ask.
- Create a new file called tools/shell_tool.py and paste the following code:
import asyncio
from tools.registry import registry
# Executes a shell command and captures stdout/stderr
async def run_shell(command: str) -> str:
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
output = stdout.decode() if stdout else ""
errors = stderr.decode() if stderr else ""
if errors:
output += f"\nSTDERR: {errors}"
return output.strip() or "(no output)"
# Self-registration: this runs at import time
registry.register(
name="run_shell",
description="Execute a shell command on the local machine and return the output.",
parameters={
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute",
}
},
"required": ["command"],
},
handler=run_shell,
)What does this code do?
The run_shell function uses Python's asyncio.create_subprocess_shell to run any command and capture its output. It returns both stdout and stderr so the agent gets full visibility.
The registry.register() call at the bottom is the self-registration pattern. The moment Python imports this file, the tool registers itself. No manual wiring needed.
- Save tools/shell_tool.py.
- Verify the tool registered itself by running:
python -c "import tools.shell_tool; from tools.registry import registry; print(len(registry.get_schemas()), 'tool(s) registered')"You should see 1 tool(s) registered printed.
Seeing 0 tools registered?
- Make sure import tools.shell_tool comes before you check registry.get_schemas(). The import triggers registration.
- Check that your shell_tool.py has the registry.register() call at module level (not inside a function or if __name__ block).
Why isn't my shell tool showing up in the registry?
Create the file, web, and memory tools
Now you will create three more tool files following the same self-registration pattern. Each file defines its handler function, then calls registry.register() at module level.
- Create a new file called tools/file_tools.py and paste the following code:
import os
from tools.registry import registry
# Read the contents of a file, with truncation for large files
async def read_file(path: str) -> str:
expanded = os.path.expanduser(path)
if not os.path.exists(expanded):
return f"Error: File not found: {path}"
with open(expanded, "r") as f:
content = f.read()
if len(content) > 10000:
content = content[:10000] + "\n... (truncated)"
return content
# Write content to a file, creating directories if needed
async def write_file(path: str, content: str) -> str:
expanded = os.path.expanduser(path)
os.makedirs(os.path.dirname(expanded) or ".", exist_ok=True)
with open(expanded, "w") as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to {path}"
registry.register(
name="read_file",
description="Read the contents of a file at the given path.",
parameters={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to read",
}
},
"required": ["path"],
},
handler=read_file,
)
registry.register(
name="write_file",
description="Write content to a file at the given path. Creates directories if needed.",
parameters={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to write to",
},
"content": {
"type": "string",
"description": "The content to write",
},
},
"required": ["path", "content"],
},
handler=write_file,
)What does this code do?
- read_file expands paths like ~/Documents, checks the file exists, and returns its content (truncated at 10,000 characters to avoid overwhelming the LLM).
- write_file creates any missing parent directories before writing. This means the agent can write to paths that do not exist yet.
- Save tools/file_tools.py.
You will not see a change yet. The next file adds another tool.
- Create a new file called tools/web_tool.py and paste the following code:
import urllib.request
import urllib.error
from tools.registry import registry
# Fetch raw text content from a URL
async def web_fetch(url: str) -> str:
try:
req = urllib.request.Request(url, headers={"User-Agent": "MyAgent/1.0"})
with urllib.request.urlopen(req, timeout=10) as response:
content = response.read().decode("utf-8", errors="replace")
if len(content) > 5000:
content = content[:5000] + "\n... (truncated)"
return content
except urllib.error.URLError as e:
return f"Error fetching URL: {e}"
registry.register(
name="web_fetch",
description="Fetch the content of a web page by URL. Returns the raw text content.",
parameters={
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to fetch",
}
},
"required": ["url"],
},
handler=web_fetch,
)What does this code do?
web_fetch uses Python's built-in urllib to grab a web page. It sets a custom User-Agent header, enforces a 10-second timeout, and truncates responses at 5,000 characters to keep token usage low.
- Save tools/web_tool.py.
You will not see a change yet. One more tool to go.
- Create a new file called tools/memory_tool.py and paste the following code:
import os
from tools.registry import registry
# Path to the persistent memory file at the project root
MEMORY_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "MEMORY.md")
# Append a fact to MEMORY.md so it persists across sessions
async def remember(fact: str) -> str:
with open(MEMORY_FILE, "a") as f:
f.write(f"\n- {fact}")
return f"Remembered: {fact}"
registry.register(
name="remember",
description="Save an important fact about the user to persistent memory. Use this when you learn something worth remembering across sessions.",
parameters={
"type": "object",
"properties": {
"fact": {
"type": "string",
"description": "The fact to remember",
}
},
"required": ["fact"],
},
handler=remember,
)What does this code do?
The remember tool appends facts to your MEMORY.md file as bullet points. Because the agent's system prompt includes MEMORY.md content, anything saved here is available in future conversations.
This is the same pattern used by Hermes Agent. File-based memory is simple but effective for a local agent.
- Save tools/memory_tool.py.
- Verify all five tools registered by running:
python -c "import tools.shell_tool; import tools.file_tools; import tools.web_tool; import tools.memory_tool; from tools.registry import registry; print(len(registry.get_schemas()), 'tool(s) registered')"You should see 5 tool(s) registered printed. That confirms all five tools self-registered at import time.
Seeing fewer than 5 tools?
- Check that each tool file has from tools.registry import registry at the top and a registry.register() call at the bottom.
- If you see an ImportError, one of your files likely has a typo in its import line. Compare against the code above.
- Remember that file_tools.py registers two tools (read_file and write_file), so 4 files produce 5 tool registrations.
Help me debug why some tools are not registering.
Why self-registration?
This pattern means adding a new tool to your agent is just one step: create a file in tools/. There is no central config file to update and no list to maintain. Hermes Agent uses this exact approach to manage 61 tools across 52 toolsets.
Your agent now has five tools ready to go. Next up, you will build the agent loop that actually calls OpenAI and decides when to use these tools.
Build the Agent Loop
You have a full tool system ready to go. Tools can run shell commands, read and write files, fetch web pages, and remember facts. Now you need a brain to decide when to use them.
The agent loop is the core decision engine. It sends the user's message to OpenAI along with tool schemas, checks if the model wants to call a tool, executes that tool, feeds the result back, and loops until the model produces a final text answer.
In this step, get ready to:
- Create the Agent class with system prompt loading from SOUL.md and MEMORY.md.
- Implement the conversation loop with OpenAI function calling and tool dispatch.
- Understand how persistent memory works through the existing remember tool.
Create the Agent class
The Agent class is responsible for managing conversation history and assembling the system prompt. It reads your agent's personality from SOUL.md and appends any stored memories from MEMORY.md every time it starts a new conversation turn.
- Create a new file called agent.py in your my-agent/ folder and add the following code:
import os
import json
from openai import AsyncOpenAI
from tools.registry import registry
class Agent:
def __init__(self):
# Connect to OpenAI using the API key from environment variables
self.client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
self.model = "gpt-4o-mini"
# Store separate conversation histories keyed by session ID
self.conversations: dict[str, list[dict]] = {}
def _load_system_prompt(self) -> str:
# Build the system prompt from SOUL.md (personality) and MEMORY.md (persistent memory)
base_dir = os.path.dirname(__file__)
soul_path = os.path.join(base_dir, "SOUL.md")
memory_path = os.path.join(base_dir, "MEMORY.md")
soul = ""
if os.path.exists(soul_path):
with open(soul_path, "r") as f:
soul = f.read()
memory = ""
if os.path.exists(memory_path):
with open(memory_path, "r") as f:
memory = f.read()
return f"{soul}\n\n## Your Memory\n{memory}"
def _get_history(self, session_id: str) -> list[dict]:
# Return existing history or create a new empty list for this session
if session_id not in self.conversations:
self.conversations[session_id] = []
return self.conversations[session_id]What does this code do?
AsyncOpenAI creates an async client for calling OpenAI's API. The agent uses gpt-4o-mini because it supports full function calling and costs very little (about $0.15 per million input tokens).
_load_system_prompt() combines your agent's personality (SOUL.md) with its persistent memory (MEMORY.md) into one system message. This mirrors how Hermes Agent uses a prompt builder to inject personality and memory on every turn.
_get_history() keeps a separate conversation history per session ID. This means your Telegram chat and CLI chat each get their own isolated history.
- Save agent.py.
- Confirm the file was created by running:
ls agent.pyYou should see agent.py listed in the output.
Don't see agent.py?
Make sure you created the file inside the my-agent/ folder, not inside tools/. The agent lives at the project root alongside SOUL.md and MEMORY.md.
help me figure out where to create agent.py
Implement the conversation loop
The conversation loop is where the magic happens. It sends the full message history to OpenAI, checks if the model wants to call any tools, executes those tools, and keeps looping until the model produces a final text response.
- In agent.py, add the run_conversation method below the _get_history method by copying this code:
async def run_conversation(self, session_id: str, user_message: str) -> str:
history = self._get_history(session_id)
history.append({"role": "user", "content": user_message})
# Build the full message list: system prompt + conversation history
messages = [
{"role": "system", "content": self._load_system_prompt()},
*history,
]
# Get tool schemas so the model knows what tools are available
tools = registry.get_schemas()
max_iterations = 10
for _ in range(max_iterations):
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools if tools else None,
)
choice = response.choices[0]
if choice.finish_reason == "tool_calls":
# The model wants to call one or more tools
assistant_message = choice.message
messages.append(assistant_message.model_dump())
# Execute each tool call and feed results back
for tool_call in assistant_message.tool_calls:
result = await registry.dispatch(
tool_call.function.name,
tool_call.function.arguments,
)
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
)
else:
# The model produced a final text response
assistant_content = choice.message.content or ""
history.append({"role": "assistant", "content": assistant_content})
return assistant_content
return "I hit my tool-call limit. Please try a simpler request."What does this code do?
This is the agent loop pattern used by production agents like Hermes Agent. Here is how it flows:
- The user's message gets appended to the session history, then the full conversation (system prompt + history) is sent to OpenAI with all available tool schemas.
- If the model responds with finish_reason == "tool_calls", the agent executes each requested tool through the registry's dispatch() method and sends the results back to the model.
- The loop repeats until the model produces a text response (no more tool calls) or hits the 10-iteration safety cap.
Why a loop instead of a single call?
A single request like "list files in my project and tell me which is largest" requires two tool calls: first run_shell to list files, then read_file to check sizes. The model needs to see the first result before deciding what to do next.
The loop gives the model multiple chances to reason and act. The 10-iteration cap prevents runaway loops if the model gets stuck.
- Save agent.py.
- Verify the file has valid syntax by running:
python -c "from agent import Agent; print('Agent class loaded successfully')" You should see Agent class loaded successfully printed in your terminal.
Seeing an import error?
- If you see ModuleNotFoundError: No module named 'openai', make sure you installed the dependencies from Step 1. Run pip install openai==2.36.0 to fix it.
- If you see a SyntaxError, check that the indentation of run_conversation is at the same level as the other methods (4 spaces inside the class).
help me debug my agent.py import error
Understand how memory persists across sessions
You might be wondering how the agent remembers things across restarts. The answer is the remember tool you created in the previous step.
How does persistent memory work?
When the user says something like "remember that my favorite language is Python", the model calls the remember tool. That tool appends the fact to MEMORY.md on disk.
On the next conversation turn (or after a restart), _load_system_prompt() re-reads MEMORY.md and injects those memories into the system prompt. The model sees everything it previously saved, giving it persistent context without a database.
✔️ Awesome, I've got everything!
Great. Double check you have saved agent.py and move on to the next step.
ⓧ I'd like to double check the full code
Here is the complete agent.py file. Compare it against yours to spot any differences.
import os
import json
from openai import AsyncOpenAI
from tools.registry import registry
class Agent:
def __init__(self):
self.client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
self.model = "gpt-4o-mini"
self.conversations: dict[str, list[dict]] = {}
def _load_system_prompt(self) -> str:
base_dir = os.path.dirname(__file__)
soul_path = os.path.join(base_dir, "SOUL.md")
memory_path = os.path.join(base_dir, "MEMORY.md")
soul = ""
if os.path.exists(soul_path):
with open(soul_path, "r") as f:
soul = f.read()
memory = ""
if os.path.exists(memory_path):
with open(memory_path, "r") as f:
memory = f.read()
return f"{soul}\n\n## Your Memory\n{memory}"
def _get_history(self, session_id: str) -> list[dict]:
if session_id not in self.conversations:
self.conversations[session_id] = []
return self.conversations[session_id]
async def run_conversation(self, session_id: str, user_message: str) -> str:
history = self._get_history(session_id)
history.append({"role": "user", "content": user_message})
messages = [
{"role": "system", "content": self._load_system_prompt()},
*history,
]
tools = registry.get_schemas()
max_iterations = 10
for _ in range(max_iterations):
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools if tools else None,
)
choice = response.choices[0]
if choice.finish_reason == "tool_calls":
assistant_message = choice.message
messages.append(assistant_message.model_dump())
for tool_call in assistant_message.tool_calls:
result = await registry.dispatch(
tool_call.function.name,
tool_call.function.arguments,
)
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
)
else:
assistant_content = choice.message.content or ""
history.append({"role": "assistant", "content": assistant_content})
return assistant_content
return "I hit my tool-call limit. Please try a simpler request."Your agent now has a working brain that can reason about tools and call them autonomously. Next up, you will build the channel adapters and gateway that connect this brain to the outside world.
Build the Channel Adapters and Gateway
Your agent can think, call tools, and remember things. But right now it has no way to receive messages from the outside world.
In this step, you'll build the messaging layer that connects your agent to the real world. You'll create channel adapters for both Telegram and your terminal, plus a gateway that routes messages from any channel into the agent loop you built in the previous step.
In this step, get ready to:
- Create an abstract Channel class and a CLI channel adapter for local testing.
- Build a Telegram channel adapter using python-telegram-bot.
- Build a Gateway class that routes messages from any channel to the agent.
Create the base channel class and CLI adapter
In Hermes Agent's architecture, every messaging platform (Telegram, Discord, Slack) implements the same interface. This means the agent doesn't need to know which platform a message came from.
Why the gateway pattern?
You might expect the agent to receive messages directly from Telegram or a terminal. The gateway pattern adds a routing layer in between.
This decouples your agent from any specific messaging platform. Adding a new channel (Discord, Slack, a web API) means writing one small adapter file. The agent code never changes.
Your channels/ directory already exists from Step 1. It needs an __init__.py to make Python treat it as a package.
- Create the empty package file by running this command:
touch channels/__init__.py- Confirm the file exists by running:
ls channels/You should see __init__.py in the output.
Now you'll define the abstract base class that all channels must implement. This guarantees every adapter has the same interface.
- Create a new file called channels/base.py and add the following code:
from abc import ABC, abstractmethod
from typing import Callable, Awaitable
class Channel(ABC):
def __init__(self, channel_id: str):
self.channel_id = channel_id
self._message_handler: Callable[[str, str, str], Awaitable[str]] | None = None
def set_message_handler(
self, handler: Callable[[str, str, str], Awaitable[str]]
):
self._message_handler = handler
@abstractmethod
async def start(self):
pass
@abstractmethod
async def send_message(self, user_id: str, text: str):
passWhat does this code do?
- Channel is an abstract base class. Any class that extends it must implement start() and send_message().
- _message_handler is a callback the gateway will inject. When a channel receives a message, it calls this handler to route it to the agent.
- The type hint Callable[[str, str, str], Awaitable[str]] means the handler takes three strings (channel_id, user_id, text) and returns a string asynchronously.
- Save channels/base.py.
You won't see a visible change yet. The CLI channel you build next will use this class.
The CLI channel is the simplest adapter. It reads from your terminal's standard input and prints responses back to standard output.
- Create a new file called channels/cli_channel.py and add the following code:
import asyncio
from channels.base import Channel
class CLIChannel(Channel):
def __init__(self):
super().__init__(channel_id="cli")
async def start(self):
print("CLI channel ready. Type your messages (Ctrl+C to quit):")
while True:
try:
user_input = await asyncio.get_event_loop().run_in_executor(
None, input, "\nYou: "
)
if not user_input.strip():
continue
if self._message_handler:
response = await self._message_handler(
self.channel_id, "cli_user", user_input
)
print(f"\nAgent: {response}")
except (EOFError, KeyboardInterrupt):
print("\nGoodbye!")
break
async def send_message(self, user_id: str, text: str):
print(f"\nAgent: {text}")What does this code do?
- run_in_executor wraps the blocking input() call so it doesn't freeze the async event loop. This lets other tasks (like the Telegram channel) keep running while waiting for terminal input.
- When the user types something, the channel calls _message_handler with the channel ID, a fixed user ID of "cli_user", and the input text.
- Ctrl+C exits the loop cleanly with a goodbye message.
- Save channels/cli_channel.py.
- Confirm both channel files exist by running:
ls channels/You should see __init__.py, base.py, and cli_channel.py in the output.
Seeing a ModuleNotFoundError?
If you test importing and see ModuleNotFoundError: No module named 'channels', make sure channels/__init__.py exists. Python needs this file to treat the folder as a package.
help me fix a ModuleNotFoundError for my channels package
Create the Telegram channel adapter
The Telegram adapter follows the same interface as the CLI channel, but connects to Telegram's servers using the python-telegram-bot library you installed earlier. It registers handlers for incoming messages and forwards them to the gateway.
- Create a new file called channels/telegram_channel.py and add the following code:
import os
from telegram import Update
from telegram.ext import (
Application,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)
from channels.base import Channel
class TelegramChannel(Channel):
def __init__(self):
super().__init__(channel_id="telegram")
self.app: Application | None = None
async def start(self):
token = os.environ.get("TELEGRAM_BOT_TOKEN", "")
if not token:
print("Warning: TELEGRAM_BOT_TOKEN not set. Telegram channel disabled.")
return
self.app = Application.builder().token(token).build()
self.app.add_handler(CommandHandler("start", self._handle_start))
self.app.add_handler(
MessageHandler(filters.TEXT & ~filters.COMMAND, self._handle_message)
)
await self.app.initialize()
await self.app.start()
await self.app.updater.start_polling(allowed_updates=Update.ALL_TYPES)
print("Telegram channel connected.")
async def _handle_start(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
await update.message.reply_text(
"Hello! I am your personal AI agent. Send me a message!"
)
async def _handle_message(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
if not update.message or not update.message.text:
return
user_id = str(update.effective_user.id)
text = update.message.text
if self._message_handler:
response = await self._message_handler(self.channel_id, user_id, text)
await update.message.reply_text(response)
async def send_message(self, user_id: str, text: str):
if self.app:
await self.app.bot.send_message(chat_id=int(user_id), text=text)
async def stop(self):
if self.app:
await self.app.updater.stop()
await self.app.stop()
await self.app.shutdown()What does this code do?
- start() checks for the TELEGRAM_BOT_TOKEN environment variable. If it's missing, the channel gracefully disables itself instead of crashing.
- Application.builder().token(token).build() creates the Telegram bot application using the builder pattern from python-telegram-bot.
- Two handlers are registered: a /start command handler (sends a welcome message), and a text message handler that forwards all non-command messages to the gateway.
- start_polling() runs in the background, continuously checking Telegram's servers for new messages.
- stop() cleanly shuts down polling and the application when your agent exits.
- Save channels/telegram_channel.py.
You won't see a change yet. The Telegram channel needs the gateway to route messages, which you'll build next.
Seeing an import error for telegram?
If your editor shows a red underline on from telegram import Update, make sure you installed python-telegram-bot in Step 1. Run pip install python-telegram-bot==22.7 if needed.
help me fix my telegram import
Create the gateway
The gateway is the central router. It holds a reference to the agent and all registered channels. When any channel receives a message, the gateway builds a unique session ID and passes the message to the agent.
- Create a new file called gateway.py in your my-agent/ folder (at the same level as agent.py) and add the following code:
from agent import Agent
from channels.base import Channel
class Gateway:
def __init__(self, agent: Agent):
self.agent = agent
self.channels: dict[str, Channel] = {}
def register_channel(self, channel: Channel):
channel.set_message_handler(self.handle_message)
self.channels[channel.channel_id] = channel
async def handle_message(
self, channel_id: str, user_id: str, text: str
) -> str:
session_id = f"{channel_id}:{user_id}"
response = await self.agent.run_conversation(session_id, text)
return response
async def start(self):
for channel in self.channels.values():
await channel.start()What does this code do?
- register_channel() injects the gateway's handle_message method as the channel's callback. This is how channels know where to send incoming messages without importing the agent directly.
- handle_message() builds a session ID like "telegram:12345" or "cli:cli_user". This gives each channel+user pair its own isolated conversation history.
- start() loops through all registered channels and starts them. Telegram begins polling in the background while the CLI blocks in the foreground.
- Save gateway.py.
- Confirm all the new files exist by running:
ls channels/ gateway.pyYou should see __init__.py, base.py, cli_channel.py, and telegram_channel.py listed under channels/, plus gateway.py at the project root.
Missing a file?
Double check you created each file in the correct location. gateway.py lives in my-agent/ (next to agent.py), while all channel files live inside my-agent/channels/.
help me find my missing files
✔️ Awesome, I've got everything!
Great. Double check you've saved all your files before moving on.
ⓧ I'd like to double check the full code
Here is the full code for each file. Compare against yours:
from abc import ABC, abstractmethod
from typing import Callable, Awaitable
class Channel(ABC):
def __init__(self, channel_id: str):
self.channel_id = channel_id
self._message_handler: Callable[[str, str, str], Awaitable[str]] | None = None
def set_message_handler(
self, handler: Callable[[str, str, str], Awaitable[str]]
):
self._message_handler = handler
@abstractmethod
async def start(self):
pass
@abstractmethod
async def send_message(self, user_id: str, text: str):
passimport asyncio
from channels.base import Channel
class CLIChannel(Channel):
def __init__(self):
super().__init__(channel_id="cli")
async def start(self):
print("CLI channel ready. Type your messages (Ctrl+C to quit):")
while True:
try:
user_input = await asyncio.get_event_loop().run_in_executor(
None, input, "\nYou: "
)
if not user_input.strip():
continue
if self._message_handler:
response = await self._message_handler(
self.channel_id, "cli_user", user_input
)
print(f"\nAgent: {response}")
except (EOFError, KeyboardInterrupt):
print("\nGoodbye!")
break
async def send_message(self, user_id: str, text: str):
print(f"\nAgent: {text}")import os
from telegram import Update
from telegram.ext import (
Application,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)
from channels.base import Channel
class TelegramChannel(Channel):
def __init__(self):
super().__init__(channel_id="telegram")
self.app: Application | None = None
async def start(self):
token = os.environ.get("TELEGRAM_BOT_TOKEN", "")
if not token:
print("Warning: TELEGRAM_BOT_TOKEN not set. Telegram channel disabled.")
return
self.app = Application.builder().token(token).build()
self.app.add_handler(CommandHandler("start", self._handle_start))
self.app.add_handler(
MessageHandler(filters.TEXT & ~filters.COMMAND, self._handle_message)
)
await self.app.initialize()
await self.app.start()
await self.app.updater.start_polling(allowed_updates=Update.ALL_TYPES)
print("Telegram channel connected.")
async def _handle_start(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
await update.message.reply_text(
"Hello! I am your personal AI agent. Send me a message!"
)
async def _handle_message(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
if not update.message or not update.message.text:
return
user_id = str(update.effective_user.id)
text = update.message.text
if self._message_handler:
response = await self._message_handler(self.channel_id, user_id, text)
await update.message.reply_text(response)
async def send_message(self, user_id: str, text: str):
if self.app:
await self.app.bot.send_message(chat_id=int(user_id), text=text)
async def stop(self):
if self.app:
await self.app.updater.stop()
await self.app.stop()
await self.app.shutdown()from agent import Agent
from channels.base import Channel
class Gateway:
def __init__(self, agent: Agent):
self.agent = agent
self.channels: dict[str, Channel] = {}
def register_channel(self, channel: Channel):
channel.set_message_handler(self.handle_message)
self.channels[channel.channel_id] = channel
async def handle_message(
self, channel_id: str, user_id: str, text: str
) -> str:
session_id = f"{channel_id}:{user_id}"
response = await self.agent.run_conversation(session_id, text)
return response
async def start(self):
for channel in self.channels.values():
await channel.start()Your messaging layer is complete. You have an abstract base class, two concrete channel adapters, and a gateway that ties everything together. Next up, you'll wire the gateway, agent, and channels into a single entry point and run your agent for real.
Wire Everything Together and Run
You have all the building blocks ready. The tool registry, the agent loop, the channel adapters, and the gateway are all built and waiting.
Now you need an entry point that boots everything up in the right order. The main.py file orchestrates the startup sequence: load config, trigger tool self-registration via imports, create the agent and gateway, and start both channels.
In this step, get ready to:
- Create the main.py entry point that wires all components together.
- Create a Telegram bot via BotFather and connect it to your agent.
- Test the agent from both CLI and Telegram with real tool calls and memory.
Create the entry point
The entry point is the file you actually run to start your agent. It needs to do things in a specific order: environment variables first (so credentials are available), then tool imports (which trigger self-registration), then component creation and startup.
- Create a new file called main.py in your my-agent/ folder and paste the following code:
import asyncio
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Import tool modules to trigger self-registration
import tools.shell_tool
import tools.file_tools
import tools.web_tool
import tools.memory_tool
from agent import Agent
from gateway import Gateway
from channels.cli_channel import CLIChannel
from channels.telegram_channel import TelegramChannel
async def main():
agent = Agent()
gateway = Gateway(agent)
# Register channels
telegram = TelegramChannel()
cli = CLIChannel()
gateway.register_channel(telegram)
gateway.register_channel(cli)
# Start Telegram in background
await telegram.start()
# Run CLI in foreground (blocks until user exits)
await cli.start()
# Cleanup
await telegram.stop()
if __name__ == "__main__":
asyncio.run(main())What does this code do?
The startup sequence is carefully ordered:
- load_dotenv() reads your .env file and makes the API keys available as environment variables.
- The four import tools.* lines look like they do nothing, but each import triggers the registry.register() calls at the bottom of those files. This is the self-registration pattern.
- Telegram starts first (non-blocking poll in the background), then CLI starts and blocks the main thread until you press Ctrl+C.
- When you exit, telegram.stop() gracefully shuts down the Telegram connection.
Why import order matters here
The tool imports come AFTER load_dotenv() but BEFORE creating the Agent. This guarantees that environment variables are ready when tools load, and that all tools are registered before the agent tries to list their schemas.
- Save main.py.
Seeing a ModuleNotFoundError for dotenv?
Make sure you installed python-dotenv back in Step 1. Run pip install python-dotenv==1.1.0 to fix it.
help me fix a ModuleNotFoundError when running main.py
Create a Telegram bot with BotFather
Your Telegram channel needs a real bot token to connect. Telegram lets you create bots instantly through a special bot called BotFather.
- Open Telegram on your phone or desktop.
- Search for @BotFather and open a chat with it.
- Send the message:
/newbot- BotFather will ask you to choose a display name. Type any name you like (e.g. My Personal Agent).
- BotFather will then ask for a username. This must end in bot (e.g. myagent_2024_bot).
- BotFather responds with your bot token. It looks like a long string of numbers, a colon, and letters.
Keep your token private
Your bot token gives full control of your bot. Never commit it to Git or share it publicly.
- Copy the token BotFather gave you.
- Open your .env file in Cursor.
- Find this line:
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-here- Replace your-telegram-bot-token-here with the actual token you copied from BotFather, so the line looks like:
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrSTUvwxYZ- Save the .env file.
Can't find BotFather?
Make sure you search for exactly @BotFather (with a capital B and F). It has a blue verification checkmark. If your Telegram search is empty, check your internet connection.
help me create a Telegram bot with BotFather
Start the agent and test it
Everything is connected. Time to start the agent and confirm that tool calling, memory, and both channels work end to end.
- Start your agent by running this command:
python main.pyYou should see two lines print in your terminal:
Telegram channel connected.
CLI channel ready. Type your messages (Ctrl+C to quit):Seeing an error on startup?
- If you see OPENAI_API_KEY errors, double-check your .env file has a valid key with no extra spaces.
- If you see Warning: TELEGRAM_BOT_TOKEN not set, your token is missing or empty in .env.
- If you see Conflict: terminated by other getUpdates request, another instance of your bot is already running. Stop it first with Ctrl+C in the other terminal.
help me fix my agent startup error
Now test the agent from the CLI. Try a tool-calling command first.
- At the You: prompt, type:
list the files in my current directoryThe agent calls the run_shell tool behind the scenes and responds with your directory listing. You should see file names like main.py, agent.py, gateway.py, and the tools/ and channels/ folders.
- Now test memory. Type:
remember that my favorite language is PythonThe agent calls the remember tool and confirms it saved the fact. This appends a line to your MEMORY.md file.
- Stop the agent by pressing Ctrl+C.
- Open MEMORY.md in Cursor. You should see a new bullet point: - my favorite language is Python.
Why check MEMORY.md directly?
This confirms persistence works. When you restart the agent, MEMORY.md gets loaded into the system prompt. The agent will know your favorite language without being told again.
- Restart the agent to verify memory persistence:
python main.py- Type:
what is my favorite programming language?The agent should respond with "Python" without needing to be told again. It read the fact from MEMORY.md on startup.
- Now test the Telegram channel. Open Telegram and find the bot you created (search for the username you chose, e.g. @myagent_2024_bot).
- Tap Start or send /start. The bot should reply with a greeting.
- Send a message like what files are in my home directory? and confirm the agent responds with a file listing.
Telegram bot not responding?
- Check your terminal. You should see Telegram channel connected. when the agent started. If you see the warning instead, your token is wrong.
- Make sure you're messaging the correct bot (the one whose token is in your .env).
- If the bot was working but stopped, your agent process may have crashed. Check the terminal for errors.
help me debug my Telegram bot connection
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py and your .env file with the real Telegram token.
ⓧ I'd like to double check the full code
Here is the complete main.py file:
import asyncio
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Import tool modules to trigger self-registration
import tools.shell_tool
import tools.file_tools
import tools.web_tool
import tools.memory_tool
from agent import Agent
from gateway import Gateway
from channels.cli_channel import CLIChannel
from channels.telegram_channel import TelegramChannel
async def main():
agent = Agent()
gateway = Gateway(agent)
# Register channels
telegram = TelegramChannel()
cli = CLIChannel()
gateway.register_channel(telegram)
gateway.register_channel(cli)
# Start Telegram in background
await telegram.start()
# Run CLI in foreground (blocks until user exits)
await cli.start()
# Cleanup
await telegram.stop()
if __name__ == "__main__":
asyncio.run(main())And your .env should look like this (with your real keys):
OPENAI_API_KEY=your-api-key-here
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-hereYour personal AI agent is alive. It routes messages through the gateway, calls tools autonomously, and remembers you across sessions. Next up, the Secret Mission extends it with proactive scheduled messages.
Secret mission
Add a Cron Scheduler for Proactive Messages
Your agent only responds when spoken to. Real AI assistants like Hermes Agent can reach out to you proactively, sending scheduled reminders and updates without being asked. In this mission, you'll build a cron scheduler that runs alongside your gateway.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely on your local machine. The only cost is OpenAI API usage, which only occurs when the agent is actively processing messages.
Resources you used:
- Project directory (my-agent/) with all Python source files.
- Python packages: openai, python-telegram-bot, python-dotenv.
- OpenAI API key (stored in .env).
- Telegram bot (created via BotFather).
- Persistent memory file (MEMORY.md).
Keep everything running
No action needed. Choose this if you plan to keep using your AI agent day to day.
- Your project files stay on your Desktop.
- Your Telegram bot stays active and ready to receive messages.
- Your OpenAI API key remains valid. You only pay when you send messages to the agent.
Pause - I'll come back to this later
Stop the running agent process to free up your terminal. Your files and configuration stay intact for next time.
- Press Ctrl+C in the terminal where the agent is running to stop the process.
- Your project files, MEMORY.md, and .env all remain on disk.
- To resume later, open a terminal in your my-agent/ folder and run python main.py again.
Delete - I don't want to use this again
Remove all project files, the Telegram bot, and revoke your API key if you no longer need it.
- Press Ctrl+C to stop the agent if it is still running.
- Delete the my-agent/ folder from your Desktop.
- Open Telegram, message @BotFather, and send /deletebot to remove your bot.
- If you created the OpenAI API key only for this project, revoke it in the OpenAI dashboard under API keys.
- Optionally uninstall the Python packages by running pip uninstall openai python-telegram-bot python-dotenv in your terminal.
Nice Work!
Nice Work!
You just built a fully functional personal AI agent from scratch. Your agent runs locally on your machine, connects to Telegram as a messaging channel, and executes real actions through a self-registering tool system.
You've learned how to:
- Build a tool registry with self-registering tools that mirror the same pattern used by Hermes Agent's 61-tool system.
- Implement an agent loop with OpenAI function calling that autonomously executes shell commands, reads and writes files, and fetches web content.
- Create a gateway with channel adapters that route messages between Telegram, CLI, and the agent using a clean async architecture.
- Added a cron scheduler so your agent can send proactive messages on Telegram at scheduled intervals.
Ready to quiz yourself?