ilia/nanobot
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

style: satisfy Ruff I001/W293 in agent package

Browse Source 
- isort: reorder imports in __init__.py and loop.py
- Strip trailing whitespace on blank lines in context.py
- Use module-level ExecToolConfig/CronService and unquoted annotations in loop.py (fix F821)

Made-with: Cursor
This commit is contained in:
tanyar09 2026-03-27 14:24:06 -04:00
parent 5292f91548
commit a2ae3f0cea
3 changed files with 80 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
nanobot/agent/__init__.py
View File

@ -1,7 +1,7 @@
"""Agent core module."""
from nanobot.agent.loop import AgentLoop
from nanobot.agent.context import ContextBuilder
from nanobot.agent.loop import AgentLoop
from nanobot.agent.memory import MemoryStore
from nanobot.agent.skills import SkillsLoader

58
nanobot/agent/context.py
View File

@ -13,43 +13,43 @@ from nanobot.agent.skills import SkillsLoader
class ContextBuilder:
"""
Builds the context (system prompt + messages) for the agent.
Assembles bootstrap files, memory, skills, and conversation history
into a coherent prompt for the LLM.
"""
BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md", "IDENTITY.md"]
def __init__(self, workspace: Path):
self.workspace = workspace
self.memory = MemoryStore(workspace)
self.skills = SkillsLoader(workspace)
def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
"""
Build the system prompt from bootstrap files, memory, and skills.
Args:
skill_names: Optional list of skills to include.
Returns:
Complete system prompt.
"""
parts = []
# Core identity
parts.append(self._get_identity())
# Bootstrap files
bootstrap = self._load_bootstrap_files()
if bootstrap:
parts.append(bootstrap)
# Memory context
memory = self.memory.get_memory_context()
if memory:
parts.append(f"# Memory\n\n{memory}")
# Skills - progressive loading
# 1. Always-loaded skills: include full content
always_skills = self.skills.get_always_skills()
@ -57,7 +57,7 @@ class ContextBuilder:
always_content = self.skills.load_skills_for_context(always_skills)
if always_content:
parts.append(f"# Active Skills\n\n{always_content}")
# 2. Available skills: only show summary (agent uses read_file to load)
skills_summary = self.skills.build_skills_summary()
if skills_summary:
@ -67,19 +67,19 @@ The following skills extend your capabilities. To use a skill, read its SKILL.md
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.
{skills_summary}""")
return "\n\n---\n\n".join(parts)
def _get_identity(self) -> str:
"""Get the core identity section."""
from datetime import datetime
import time as _time
from datetime import datetime
now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
tz = _time.strftime("%Z") or "UTC"
workspace_path = str(self.workspace.expanduser().resolve())
system = platform.system()
runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}"
return f"""# nanobot 🐈
You are nanobot, a helpful AI assistant. You have access to tools that allow you to:
@ -120,24 +120,20 @@ Always be helpful, accurate, and concise. Before calling tools, briefly tell the
When remembering something important, write to {workspace_path}/memory/MEMORY.md
To recall past events, grep {workspace_path}/memory/HISTORY.md
IMPORTANT: For owner-initiated email queries about the mailbox itself (e.g. "what's my latest email?", "list unread emails", "search my inbox"), prefer the read_emails tool over any other method. NEVER use exec() with mail/tail/awk commands or read_file() on /var/mail - those will not work. The read_emails tool is the only way to access the mailbox via IMAP.
IMPORTANT: For email queries (latest email, email sender, inbox, etc.), ALWAYS use the read_emails tool. NEVER use exec() with mail/tail/awk commands or read_file() on /var/mail - those will not work. The read_emails tool is the only way to access emails."""
When you are replying on the email channel to an external sender, treat the incoming "Email received. From: ... Subject: ... Date: ...\\n\\n<body>" content as their message and write a direct, helpful reply. Do NOT call read_emails just because the message is an email; only call read_emails if the human explicitly asks you to inspect or search the mailbox (e.g. "check my inbox", "find an email", etc.).
**Tool failures:** If a tool result says authentication failed, API access failed, or includes a tag like [NO_CALENDAR_DATA], you have no real data from that integration—never fabricate meetings, mailbox contents, or file contents. Say access failed and what to fix. For other tool errors (e.g. missing parameters), correct the call or explain the error without inventing facts."""
def _load_bootstrap_files(self) -> str:
"""Load all bootstrap files from workspace."""
parts = []
for filename in self.BOOTSTRAP_FILES:
file_path = self.workspace / filename
if file_path.exists():
content = file_path.read_text(encoding="utf-8")
parts.append(f"## {filename}\n\n{content}")
return "\n\n".join(parts) if parts else ""
def build_messages(
self,
history: list[dict[str, Any]],
@ -182,7 +178,7 @@ When you are replying on the email channel to an external sender, treat the inco
"""Build user message content with optional base64-encoded images."""
if not media:
return text
images = []
for path in media:
p = Path(path)
@ -191,11 +187,11 @@ When you are replying on the email channel to an external sender, treat the inco
continue
b64 = base64.b64encode(p.read_bytes()).decode()
images.append({"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}})
if not images:
return text
return images + [{"type": "text", "text": text}]
def add_tool_result(
self,
messages: list[dict[str, Any]],
@ -205,13 +201,13 @@ When you are replying on the email channel to an external sender, treat the inco
) -> list[dict[str, Any]]:
"""
Add a tool result to the message list.
Args:
messages: Current message list.
tool_call_id: ID of the tool call.
tool_name: Name of the tool.
result: Tool execution result.
Returns:
Updated message list.
"""
@ -222,7 +218,7 @@ When you are replying on the email channel to an external sender, treat the inco
"content": result
})
return messages
def add_assistant_message(
self,
messages: list[dict[str, Any]],
@ -232,13 +228,13 @@ When you are replying on the email channel to an external sender, treat the inco
) -> list[dict[str, Any]]:
"""
Add an assistant message to the message list.
Args:
messages: Current message list.
content: Message content.
tool_calls: Optional tool calls.
reasoning_content: Thinking output (Kimi, DeepSeek-R1, etc.).
Returns:
Updated message list.
"""

104
nanobot/agent/loop.py
View File

@ -1,28 +1,30 @@
"""Agent loop: the core processing engine."""
import asyncio
from contextlib import AsyncExitStack
import json
import json_repair
from pathlib import Path
import re
from typing import Any, Awaitable, Callable
from contextlib import AsyncExitStack
from pathlib import Path
from typing import Awaitable, Callable
import json_repair
from loguru import logger
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMProvider
from nanobot.agent.context import ContextBuilder
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFileTool, ListDirTool
from nanobot.agent.tools.shell import ExecTool
from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
from nanobot.agent.tools.message import MessageTool
from nanobot.agent.tools.spawn import SpawnTool
from nanobot.agent.tools.cron import CronTool
from nanobot.agent.memory import MemoryStore
from nanobot.agent.subagent import SubagentManager
from nanobot.agent.tools.cron import CronTool
from nanobot.agent.tools.filesystem import EditFileTool, ListDirTool, ReadFileTool, WriteFileTool
from nanobot.agent.tools.message import MessageTool
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.shell import ExecTool
from nanobot.agent.tools.spawn import SpawnTool
from nanobot.agent.tools.web import WebFetchTool, WebSearchTool
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ExecToolConfig
from nanobot.cron.service import CronService
from nanobot.providers.base import LLMProvider
from nanobot.session.manager import Session, SessionManager
@ -49,14 +51,12 @@ class AgentLoop:
max_tokens: int = 4096,
memory_window: int = 50,
brave_api_key: str | None = None,
exec_config: "ExecToolConfig | None" = None,
cron_service: "CronService | None" = None,
exec_config: ExecToolConfig | None = None,
cron_service: CronService | None = None,
restrict_to_workspace: bool = False,
session_manager: SessionManager | None = None,
mcp_servers: dict | None = None,
):
from nanobot.config.schema import ExecToolConfig
from nanobot.cron.service import CronService
self.bus = bus
self.provider = provider
self.workspace = workspace
@ -84,13 +84,13 @@ class AgentLoop:
exec_config=self.exec_config,
restrict_to_workspace=restrict_to_workspace,
)
self._running = False
self._mcp_servers = mcp_servers or {}
self._mcp_stack: AsyncExitStack | None = None
self._mcp_connected = False
self._register_default_tools()
def _register_default_tools(self) -> None:
"""Register the default set of tools."""
# File tools (restrict to workspace if configured)
@ -99,30 +99,30 @@ class AgentLoop:
self.tools.register(WriteFileTool(allowed_dir=allowed_dir))
self.tools.register(EditFileTool(allowed_dir=allowed_dir))
self.tools.register(ListDirTool(allowed_dir=allowed_dir))
# Shell tool
self.tools.register(ExecTool(
working_dir=str(self.workspace),
timeout=self.exec_config.timeout,
restrict_to_workspace=self.restrict_to_workspace,
))
# Web tools
self.tools.register(WebSearchTool(api_key=self.brave_api_key))
self.tools.register(WebFetchTool())
# Message tool
message_tool = MessageTool(send_callback=self.bus.publish_outbound)
self.tools.register(message_tool)
# Spawn tool (for subagents)
spawn_tool = SpawnTool(manager=self.subagents)
self.tools.register(spawn_tool)
# Cron tool (for scheduling)
if self.cron_service:
self.tools.register(CronTool(self.cron_service))
# Email tool (if email channel is configured)
try:
from nanobot.agent.tools.email import EmailTool
@ -137,7 +137,7 @@ class AgentLoop:
except Exception as e:
logger.warning(f"Email tool not available: {e}")
# Email tool not available or not configured - silently skip
# Calendar tool (if calendar is configured)
try:
from nanobot.agent.tools.calendar import CalendarTool
@ -152,7 +152,7 @@ class AgentLoop:
except Exception as e:
logger.warning(f"Calendar tool not available: {e}")
# Calendar tool not available or not configured - silently skip
async def _connect_mcp(self) -> None:
"""Connect to configured MCP servers (one-time, lazy)."""
if self._mcp_connected or not self._mcp_servers:
@ -231,7 +231,7 @@ class AgentLoop:
)
logger.debug(f"LLM provider returned response, has_tool_calls={response.has_tool_calls}")
except asyncio.TimeoutError:
logger.error(f"LLM provider call timed out after 120 seconds")
logger.error("LLM provider call timed out after 120 seconds")
return "Error: Request timed out. The LLM provider may be slow or unresponsive.", tools_used
except Exception as e:
logger.error(f"LLM provider error: {e}")
@ -272,10 +272,10 @@ class AgentLoop:
final_content = self._strip_think(response.content)
logger.info(f"Final response generated. Content length: {len(final_content) if final_content else 0}")
break
if final_content is None and iteration >= self.max_iterations:
logger.warning(f"Max iterations ({self.max_iterations}) reached without final response. Last tool calls: {tools_used[-3:] if len(tools_used) >= 3 else tools_used}")
return final_content, tools_used
async def run(self) -> None:
@ -303,7 +303,7 @@ class AgentLoop:
))
except asyncio.TimeoutError:
continue
async def close_mcp(self) -> None:
"""Close MCP connections."""
if self._mcp_stack:
@ -317,7 +317,7 @@ class AgentLoop:
"""Stop the agent loop."""
self._running = False
logger.info("Agent loop stopping")
async def _process_message(
self,
msg: InboundMessage,
@ -326,25 +326,25 @@ class AgentLoop:
) -> OutboundMessage | None:
"""
Process a single inbound message.
Args:
msg: The inbound message to process.
session_key: Override session key (used by process_direct).
on_progress: Optional callback for intermediate output (defaults to bus publish).
Returns:
The response message, or None if no response needed.
"""
# System messages route back via chat_id ("channel:chat_id")
if msg.channel == "system":
return await self._process_system_message(msg)
preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
logger.info(f"Processing message from {msg.channel}:{msg.sender_id}: {preview}")
key = session_key or msg.session_key
session = self.sessions.get_or_create(key)
# Handle slash commands
cmd = msg.content.strip().lower()
if cmd == "/new":
@ -365,7 +365,7 @@ class AgentLoop:
if cmd == "/help":
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
content="🐈 nanobot commands:\n/new — Start a new conversation\n/help — Show available commands")
# Skip memory consolidation for CLI mode to avoid blocking/hanging
# Memory consolidation can be slow and CLI users want fast responses
if len(session.messages) > self.memory_window and msg.channel != "cli":
@ -406,31 +406,31 @@ class AgentLoop:
if final_content is None:
final_content = "I've completed processing but have no response to give."
preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
session.add_message("user", msg.content)
session.add_message("assistant", final_content,
tools_used=tools_used if tools_used else None)
self.sessions.save(session)
return OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content=final_content,
metadata=msg.metadata or {}, # Pass through for channel-specific needs (e.g. Slack thread_ts)
)
async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
"""
Process a system message (e.g., subagent announce).
The chat_id field contains "original_channel:original_chat_id" to route
the response back to the correct destination.
"""
logger.info(f"Processing system message from {msg.sender_id}")
# Parse origin from chat_id (format: "channel:chat_id")
if ":" in msg.chat_id:
parts = msg.chat_id.split(":", 1)
@ -440,7 +440,7 @@ class AgentLoop:
# Fallback
origin_channel = "cli"
origin_chat_id = msg.chat_id
session_key = f"{origin_channel}:{origin_chat_id}"
session = self.sessions.get_or_create(session_key)
self._set_tool_context(origin_channel, origin_chat_id)
@ -454,17 +454,17 @@ class AgentLoop:
if final_content is None:
final_content = "Background task completed."
session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
session.add_message("assistant", final_content)
self.sessions.save(session)
return OutboundMessage(
channel=origin_channel,
chat_id=origin_chat_id,
content=final_content
)
async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
"""Consolidate old messages into MEMORY.md + HISTORY.md.
@ -570,14 +570,14 @@ Respond with ONLY valid JSON, no markdown fences."""
) -> str:
"""
Process a message directly (for CLI or cron usage).
Args:
content: The message content.
session_key: Session identifier (overrides channel:chat_id for session lookup).
channel: Source channel (for tool context routing).
chat_id: Source chat ID (for tool context routing).
on_progress: Optional callback for intermediate output.
Returns:
The agent's response.
"""
@ -588,6 +588,6 @@ Respond with ONLY valid JSON, no markdown fences."""
chat_id=chat_id,
content=content
)
response = await self._process_message(msg, session_key=session_key, on_progress=on_progress)
return response.content if response else ""