Build a Nanobot-Style AI Agent in Google Colab with Tool Calling, Session Memory, Skills, and MCP Servers
In this tutorial, we construct a light-weight private AI agent impressed by the core structure of nanobot, whereas protecting each half comprehensible and runnable in Google Colab. We begin from the supplier abstraction, then transfer via device registration, session reminiscence, lifecycle hooks, abilities, and an MCP-style device server. As we progress, we don’t simply use an exterior agent framework; we recreate the core constructing blocks ourselves so we are able to clearly see how messages, instruments, reminiscence, and mannequin responses work collectively inside a sensible agent loop.
Building the Provider Abstraction and Mock LLM
import subprocess, sys
def _pip_install(*pkgs):
attempt:
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], test=True)
besides Exception as e:
print(f"(pip set up skipped/failed for {pkgs}: {e})")
_HAVE_OPENAI = False
attempt:
import openai
_HAVE_OPENAI = True
besides Exception:
_pip_install("openai>=1.0.0")
attempt:
import openai
_HAVE_OPENAI = True
besides Exception:
_HAVE_OPENAI = False
attempt:
import nest_asyncio
nest_asyncio.apply()
besides Exception:
attempt:
_pip_install("nest_asyncio")
import nest_asyncio
nest_asyncio.apply()
besides Exception:
cross
import os
import re
import json
import time
import math
import asyncio
import examine
import textwrap
import contextlib
import io
from dataclasses import dataclass, discipline
from typing import Any, Callable, Optional, Awaitable, get_type_hints
def banner(title: str) -> None:
line = "═" * 78
print(f"n{line}n {title}n{line}")
@dataclass
class ToolName:
"""A normalized request from the mannequin to run one device."""
id: str
title: str
arguments: dict
@dataclass
class Usage:
prompt_tokens: int = 0
completion_tokens: int = 0
@property
def whole(self) -> int:
return self.prompt_tokens + self.completion_tokens
@dataclass
class LLMResponse:
"""The single form each supplier should return."""
content material: Optional[str]
tool_calls: checklist[ToolCall] = discipline(default_factory=checklist)
finish_reason: str = "cease"
utilization: Usage = discipline(default_factory=Usage)
class Provider:
"""Base class. A supplier turns (messages, instruments) into an LLMResponse."""
title = "base"
async def full(self, messages: checklist[dict], instruments: checklist[dict]) -> LLMResponse:
increase NotImplementedError
class OpenAICompatibleProvider(Provider):
"""
Works with OpenAI and each OpenAI-compatible gateway (OpenRouter, DeepSeek,
Together, vLLM, LM Studio, Ollama's /v1, ...). This mirrors how nanobot speaks
to most suppliers beneath the hood.
"""
title = "openai-compatible"
def __init__(self, api_key: str, mannequin: str, base_url: Optional[str] = None):
from openai import AsyncOpenAI
self.mannequin = mannequin
self.shopper = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def full(self, messages: checklist[dict], instruments: checklist[dict]) -> LLMResponse:
kwargs: dict[str, Any] = {"mannequin": self.mannequin, "messages": messages}
if instruments:
kwargs["tools"] = instruments
kwargs["tool_choice"] = "auto"
resp = await self.shopper.chat.completions.create(**kwargs)
alternative = resp.decisions[0]
msg = alternative.message
calls: checklist[ToolCall] = []
for tc in (msg.tool_calls or []):
attempt:
args = json.hundreds(tc.operate.arguments or "{}")
besides json.JSONDecodeError:
args = {"_raw": tc.operate.arguments}
calls.append(ToolName(id=tc.id, title=tc.operate.title, arguments=args))
utilization = Usage(
prompt_tokens=getattr(resp.utilization, "prompt_tokens", 0) or 0,
completion_tokens=getattr(resp.utilization, "completion_tokens", 0) or 0,
)
return LLMResponse(
content material=msg.content material,
tool_calls=calls,
finish_reason=alternative.finish_reason or "cease",
utilization=utilization,
)
class MockProvider(Provider):
"""
A deterministic, rule-based "LLM" so this complete tutorial runs with NO API key
and NO community — letting you watch the agent loop, device calls, and reminiscence work.
It imitates the ONE factor that issues for the loop: deciding to emit a device name
(in the precise normalized form a actual mannequin would) and then, as soon as device outcomes
come again, producing a ultimate natural-language reply. The agent loop can't inform
it aside from OpenAI — that is the entire level of the supplier contract.
"""
title = "mock"
def __init__(self, mannequin: str = "mock-1"):
self.mannequin = mannequin
@staticmethod
def _last_user_text(messages: checklist[dict]) -> str:
for m in reversed(messages):
if m.get("position") == "consumer":
c = m.get("content material")
return c if isinstance(c, str) else json.dumps(c)
return ""
@staticmethod
def _already_called(messages: checklist[dict], tool_name: str) -> bool:
for m in messages:
if m.get("position") == "assistant" and m.get("tool_calls"):
for tc in m["tool_calls"]:
if tc["function"]["name"] == tool_name:
return True
return False
@staticmethod
def _extract_math(textual content: str) -> str:
"""Pull the primary math-looking chunk out of a sentence (mock-only helper)."""
t = re.sub(r"sq. roots? of (d+(?:.d+)?)", r"sqrt(1)", textual content)
t = t.change("^", "**")
sample = (r"(?:sqrt(d+(?:.d+)?)|d+(?:.d+)?)"
r"(?:s*(?:**|[+-*/])s*(?:sqrt(d+(?:.d+)?)|d+(?:.d+)?))*")
m = re.search(sample, t)
return m.group(0).strip() if m else t.strip()
@staticmethod
def _scan_memory(messages: checklist[dict]) -> tuple[Optional[str], Optional[str]]:
"""Read again easy details from prior USER turns — proves session reminiscence is
truly being fed to the mannequin (mock-only comfort)."""
title = love = None
for m in messages:
if m.get("position") == "consumer" and isinstance(m.get("content material"), str):
tx = m["content"].decrease()
nm = re.search(r"my title is (w+)", tx)
if nm:
title = nm.group(1).title()
lv = re.search(r"i (?:love|like) (w+)", tx)
if lv:
love = lv.group(1).title()
return title, love
async def full(self, messages: checklist[dict], instruments: checklist[dict]) -> LLMResponse:
await asyncio.sleep(0)
consumer = self._last_user_text(messages).decrease()
tool_names = {t["function"]["name"] for t in instruments}
utilization = Usage(prompt_tokens=sum(len(str(m)) for m in messages) // 4, completion_tokens=12)
def name(title, args):
return LLMResponse(
content material=None,
tool_calls=[ToolCall(id=f"call_{name}_{int(time.time()*1000)%100000}",
name=name, arguments=args)],
finish_reason="tool_calls",
utilization=utilization,
)
has_digit = bool(re.search(r"d", consumer))
wants_math = has_digit and (
bool(re.search(r"[+-*/^]", consumer)) or "sqrt" in consumer
or "sq. root" in consumer
or any(w in consumer for w in ["calculate", "compute", "evaluate", "what is", "what's"]))
if "calculator" in tool_names and wants_math and not self._already_called(messages, "calculator"):
return name("calculator", {"expression": self._extract_math(consumer)})
if "get_current_time" in tool_names and not self._already_called(messages, "get_current_time"):
if any(w in consumer for w in ["time", "date", "today", "now", "o'clock"]):
tz = "UTC"
m = re.search(r"in ([a-zA-Z_/ ]+)", consumer)
if m:
cand = m.group(1).strip().title().change(" ", "_")
tz = {"Tokyo": "Asia/Tokyo", "Delhi": "Asia/Kolkata",
"New_York": "America/New_York", "London": "Europe/London"}.get(cand, cand)
return name("get_current_time", {"timezone": tz})
if "remember_fact" in tool_names and not self._already_called(messages, "remember_fact"):
m = re.search(r"my favourite (?:programming )?language is (w+)", consumer)
if m:
return name("remember_fact", {"key": "favorite_language", "worth": m.group(1)})
if "recall_fact" in tool_names and not self._already_called(messages, "recall_fact"):
if any(w in consumer for w in ["my favorite", "do you remember", "recall", "what did i tell"]):
key = "favorite_language" if "language" in consumer else "notice"
return name("recall_fact", {"key": key})
if "run_python" in tool_names and not self._already_called(messages, "run_python"):
py_kw = any(w in consumer for w in ["fibonacci", "prime", "factorial", "simulate"])
py_action = "python" in consumer and any(
w in consumer for w in ["run", "write", "code", "print", "execute", "snippet"])
if py_kw or py_action:
if "fibonacci" in consumer:
code = ("def fib(n):n a,b=0,1n out=[]n"
" for _ in vary(n):n out.append(a); a,b=b,a+bn return outn"
"print(fib(12))")
elif "prime" in consumer:
code = ("primes=[n for n in range(2,50) "
"if all(n%d for d in range(2,int(n**0.5)+1))]nprint(primes)")
elif "factorial" in consumer:
code = "import math; print(math.factorial(10))"
else:
code = "print(sum(vary(1,101)))"
return name("run_python", {"code": code})
if "web_search" in tool_names and not self._already_called(messages, "web_search"):
if any(w in consumer for w in ["search", "look up", "latest", "news about", "find information"]):
return name("web_search", {"question": self._last_user_text(messages)})
if any(p in consumer for p in ["my name", "who am i", "what do i love", "what i love"]):
title, love = self._scan_memory(messages)
bits = []
if title:
bits.append(f"your title is {title}")
if love:
bits.append(f"you're keen on {love}")
if bits:
return LLMResponse(content material="From our dialog, " + " and ".be a part of(bits) + ".",
tool_calls=[], finish_reason="cease", utilization=utilization)
tool_outputs = [m["content"] for m in messages if m.get("position") == "device"]
if tool_outputs:
joined = " ".be a part of(tool_outputs)
reply = f"Based on the device outcomes, this is what I discovered: {joined}"
elif any(w in consumer for w in ["hello", "hi", "hey"]):
reply = "Hello! I'm a mock nanobot agent. Ask me to calculate, inform time, run Python, or keep in mind issues."
else:
reply = ("[mock LLM] I'd usually purpose about this with a actual mannequin. "
"Set NANOBOT_API_KEY to make use of a reside LLM. For now, attempt prompts with math, "
"time, Python, or reminiscence so you may see the device loop hearth.")
return LLMResponse(content material=reply, tool_calls=[], finish_reason="cease", utilization=utilization)
We arrange the setting, set up non-obligatory dependencies, and put together the imports wanted for the complete tutorial. We outline a supplier abstraction that permits the agent to work with both a actual OpenAI-compatible mannequin or a deterministic mock supplier. We additionally construct the normalized response constructions so the remainder of the agent loop can work independently of the backend mannequin.
Creating the Tool Registry and Token-Budgeted Memory
_PYTYPE_TO_JSON = {str: "string", int: "integer", float: "quantity", bool: "boolean",
checklist: "array", dict: "object"}
@dataclass
class Tool:
title: str
description: str
parameters: dict
func: Callable
is_async: bool
def spec(self) -> dict:
"""OpenAI-style device spec the mannequin sees."""
return {"kind": "operate",
"operate": {"title": self.title,
"description": self.description,
"parameters": self.parameters}}
async def __call__(self, **kwargs) -> str:
attempt:
consequence = self.func(**kwargs)
if examine.isawaitable(consequence):
consequence = await consequence
return consequence if isinstance(consequence, str) else json.dumps(consequence, default=str)
besides Exception as e:
return f"ERROR operating device '{self.title}': {kind(e).__name__}: {e}"
def device(func: Optional[Callable] = None, *, title: Optional[str] = None):
"""
Decorator that turns a plain operate into a Tool, deriving the JSON schema from
kind hints and the primary line of the docstring. Param descriptions may be added
with a easy 'param: description' block in the docstring.
Example:
@device
def calculator(expression: str) -> str:
'''Evaluate a math expression and return the consequence.
expression: a math expression like "2 + 2 * 3" or "sqrt(16)"'''
...
"""
def make(f: Callable) -> Tool:
hints = get_type_hints(f)
sig = examine.signature(f)
doc = examine.getdoc(f) or ""
abstract = doc.break up("n", 1)[0].strip() or f.__name__
param_docs: dict[str, str] = {}
for line in doc.splitlines()[1:]:
m = re.match(r"s*(w+)s*:s*(.+)", line)
if m and m.group(1) in sig.parameters:
param_docs[m.group(1)] = m.group(2).strip()
props, required = {}, []
for pname, p in sig.parameters.objects():
if pname == "self":
proceed
jtype = _PYTYPE_TO_JSON.get(hints.get(pname, str), "string")
schema = {"kind": jtype}
if pname in param_docs:
schema["description"] = param_docs[pname]
props[pname] = schema
if p.default is examine.Parameter.empty:
required.append(pname)
parameters = {"kind": "object", "properties": props, "required": required}
return Tool(title=title or f.__name__, description=abstract,
parameters=parameters, func=f, is_async=examine.iscoroutinefunction(f))
return make(func) if func else make
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def add(self, t: Tool) -> None:
self._tools[t.name] = t
def add_function(self, f: Callable) -> None:
self.add(device(f))
def get(self, title: str) -> Optional[Tool]:
return self._tools.get(title)
def specs(self) -> checklist[dict]:
return [t.spec() for t in self._tools.values()]
def names(self) -> checklist[str]:
return checklist(self._tools)
@device
def calculator(expression: str) -> str:
"""Evaluate an arithmetic expression and return the numeric consequence.
expression: a math expression, e.g. '2 + 2 * 3', 'sqrt(16)', '2 ** 10'"""
allowed = {okay: getattr(math, okay) for okay in dir(math) if not okay.startswith("_")}
allowed.replace({"abs": abs, "spherical": spherical, "min": min, "max": max, "sqrt": math.sqrt})
expr = expression.change("^", "**")
worth = eval(expr, {"__builtins__": {}}, allowed)
return f"{expression} = {worth}"
@device
def get_current_time(timezone: str = "UTC") -> str:
"""Return the present date and time for an IANA timezone title.
timezone: IANA tz like 'UTC', 'Asia/Tokyo', 'Asia/Kolkata', 'America/New_York'"""
from datetime import datetime
attempt:
from zoneinfo import ZoneInfo
now = datetime.now(ZoneInfo(timezone))
besides Exception:
from datetime import timezone as _tz
now = datetime.now(_tz.utc)
timezone = "UTC (fallback)"
return f"Current time in {timezone}: "
@device
def run_python(code: str) -> str:
"""Execute a brief Python snippet in a restricted namespace and return its stdout.
code: Python supply code to run; use print(...) to supply output"""
safe_builtins = {"print": print, "vary": vary, "len": len, "sum": sum, "min": min,
"max": max, "abs": abs, "sorted": sorted, "enumerate": enumerate,
"checklist": checklist, "dict": dict, "set": set, "str": str, "int": int,
"float": float, "bool": bool, "map": map, "filter": filter,
"zip": zip, "all": all, "any": any, "spherical": spherical}
import math as _m
g = {"__builtins__": safe_builtins, "math": _m}
buf = io.StringIO()
attempt:
with contextlib.redirect_stdout(buf):
exec(code, g, {})
out = buf.getvalue().strip()
return f"stdout:n{out}" if out else "(ran efficiently, no stdout)"
besides Exception as e:
return f"Python error: {kind(e).__name__}: {e}"
@device
def web_search(question: str) -> str:
"""Search the net for a question and return brief consequence snippets (STUB).
question: the search question string"""
return (f"[stub results for '{query}'] (1) Overview article. (2) Official docs. "
f"(3) Recent dialogue. Swap web_search's physique for a actual API in manufacturing.")
def estimate_tokens(messages: checklist[dict]) -> int:
"""Rough token estimate (~4 chars/token) — ok for budgeting demos."""
chars = 0
for m in messages:
chars += len(str(m.get("content material") or ""))
for tc in (m.get("tool_calls") or []):
chars += len(json.dumps(tc))
return max(1, chars // 4)
class Memory:
def __init__(self, token_budget: int = 3000):
self.token_budget = token_budget
self._sessions: dict[str, list[dict]] = {}
def historical past(self, session_key: str) -> checklist[dict]:
return self._sessions.setdefault(session_key, [])
def append(self, session_key: str, message: dict) -> None:
self.historical past(session_key).append(message)
def prolong(self, session_key: str, messages: checklist[dict]) -> None:
self.historical past(session_key).prolong(messages)
def compact(self, session_key: str) -> int:
"""Drop oldest messages till beneath the token funds. Returns #dropped.
Keeps tool-call/tool-result pairs constant by trimming from the entrance in
entire turns. (nanobot additionally summarizes; we maintain it to trimming for readability.)"""
hist = self.historical past(session_key)
dropped = 0
whereas estimate_tokens(hist) > self.token_budget and len(hist) > 2:
hist.pop(0)
dropped += 1
whereas hist and hist[0].get("position") == "device":
hist.pop(0); dropped += 1
return dropped
We create a device system that permits extraordinary Python features to grow to be callable agent instruments. We use kind hints and docstrings to mechanically generate JSON-style device schemas, which makes the framework simpler to increase. We additionally add sensible offline instruments equivalent to a calculator, a time lookup device, a Python execution device, a net search stub, and token-budgeted reminiscence.
Implementing Lifecycle Hooks, Skills, and the Agent Loop
@dataclass
class AgentHookContext:
iteration: int = 0
messages: checklist[dict] = discipline(default_factory=checklist)
response: Optional[LLMResponse] = None
utilization: Usage = discipline(default_factory=Usage)
tool_calls: checklist[ToolCall] = discipline(default_factory=checklist)
tool_results: checklist[str] = discipline(default_factory=checklist)
final_content: Optional[str] = None
stop_reason: Optional[str] = None
error: Optional[Exception] = None
class AgentHook:
"""Subclass and override what you want. All async strategies are best-effort and
remoted (one failing hook will not crash the agent)."""
def wants_streaming(self) -> bool:
return False
async def before_iteration(self, context: AgentHookContext) -> None: ...
async def on_stream(self, context: AgentHookContext, delta: str) -> None: ...
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None: ...
async def before_execute_tools(self, context: AgentHookContext) -> None: ...
async def after_iteration(self, context: AgentHookContext) -> None: ...
def finalize_content(self, context: AgentHookContext, content material: str) -> str:
return content material
async def _fan_out(hooks: checklist[AgentHook], methodology: str, *args, **kwargs) -> None:
for h in hooks:
attempt:
await getattr(h, methodology)(*args, **kwargs)
besides Exception as e:
print(f" (hook {kind(h).__name__}.{methodology} error: {e})")
@dataclass
class Skill:
title: str
description: str
directions: str = ""
instruments: checklist[Tool] = discipline(default_factory=checklist)
class MCPServer:
"""Minimal stand-in for an MCP server exposing named instruments."""
def __init__(self, title: str):
self.title = title
self._impls: dict[str, dict] = {}
def register(self, title: str, description: str, parameters: dict, handler: Callable):
self._impls[name] = {"description": description, "parameters": parameters, "handler": handler}
def list_tools(self) -> checklist[dict]:
return [{"name": n, "description": v["description"], "parameters": v["parameters"]}
for n, v in self._impls.objects()]
async def call_tool(self, title: str, arguments: dict) -> str:
impl = self._impls[name]
res = impl["handler"](**arguments)
if examine.isawaitable(res):
res = await res
return res if isinstance(res, str) else json.dumps(res, default=str)
def mcp_tools(server: MCPServer) -> checklist[Tool]:
"""Adapt each device on an MCP server into our native Tool objects."""
out: checklist[Tool] = []
for spec in server.list_tools():
nm = spec["name"]
async def _runner(_nm=nm, **kwargs):
return await server.call_tool(_nm, kwargs)
out.append(Tool(title=f"{server.title}__{nm}",
description=f"[MCP:{server.name}] {spec['description']}",
parameters=spec["parameters"], func=_runner, is_async=True))
return out
@dataclass
class RunConsequence:
content material: str
tools_used: checklist[str] = discipline(default_factory=checklist)
iterations: int = 0
utilization: Usage = discipline(default_factory=Usage)
messages: checklist[dict] = discipline(default_factory=checklist)
class Agent:
def __init__(self, supplier: Provider, registry: ToolRegistry, reminiscence: Memory,
system_prompt: str, max_iterations: int = 6, verbose: bool = True):
self.supplier = supplier
self.registry = registry
self.reminiscence = reminiscence
self.system_prompt = system_prompt
self.max_iterations = max_iterations
self.verbose = verbose
def _log(self, *a):
if self.verbose:
print(*a)
async def run(self, user_message: str, *, session_key: str = "default",
hooks: Optional[list[AgentHook]] = None,
extra_instructions: str = "") -> RunConsequence:
hooks = hooks or []
system = self.system_prompt
if extra_instructions:
system += "nn" + extra_instructions
self.reminiscence.append(session_key, {"position": "consumer", "content material": user_message})
dropped = self.reminiscence.compact(session_key)
if dropped:
self._log(f" · reminiscence compaction dropped {dropped} previous message(s)")
messages = [{"role": "system", "content": system}, *self.memory.history(session_key)]
ctx = AgentHookContext(messages=messages)
tools_used: checklist[str] = []
whole = Usage()
final_text = ""
for i in vary(1, self.max_iterations + 1):
ctx.iteration = i
ctx.messages = messages
await _fan_out(hooks, "before_iteration", ctx)
response = await self.supplier.full(messages, self.registry.specs())
ctx.response = response
whole.prompt_tokens += response.utilization.prompt_tokens
whole.completion_tokens += response.utilization.completion_tokens
ctx.utilization = whole
if response.tool_calls:
ctx.tool_calls = response.tool_calls
self._log(f" [iter {i}] mannequin requested {len(response.tool_calls)} device name(s)")
messages.append({
"position": "assistant",
"content material": response.content material,
"tool_calls": [{"id": tc.id, "type": "function",
"function": {"name": tc.name,
"arguments": json.dumps(tc.arguments)}}
for tc in response.tool_calls],
})
await _fan_out(hooks, "before_execute_tools", ctx)
outcomes: checklist[str] = []
for tc in response.tool_calls:
t = self.registry.get(tc.title)
if t is None:
consequence = f"ERROR: unknown device '{tc.title}'"
else:
consequence = await t(**tc.arguments)
tools_used.append(tc.title)
outcomes.append(consequence)
self._log(f" ↳ {tc.title}({tc.arguments}) -> {consequence[:120]}")
messages.append({"position": "device", "tool_call_id": tc.id,
"content material": consequence})
ctx.tool_results = outcomes
await _fan_out(hooks, "after_iteration", ctx)
proceed
final_text = response.content material or ""
for h in hooks:
attempt:
final_text = h.finalize_content(ctx, final_text)
besides Exception as e:
print(f" (hook {kind(h).__name__}.finalize_content error: {e})")
ctx.final_content = final_text
ctx.stop_reason = response.finish_reason
await _fan_out(hooks, "after_iteration", ctx)
self.reminiscence.append(session_key, {"position": "assistant", "content material": final_text})
break
else:
final_text = "(stopped: hit max_iterations with out a ultimate reply)"
return RunConsequence(content material=final_text, tools_used=tools_used,
iterations=ctx.iteration, utilization=whole,
messages=checklist(messages))
We implement the lifecycle hooks, ability construction, MCP-style server adapter, and the primary agent loop. We use hooks to watch or modify the agent’s habits with out altering the core runtime. We then run the central loop the place the mannequin receives messages, requests instruments when wanted, consumes device outcomes, and lastly returns a plain-text reply.
Wrapping the Agent in a Nanobot SDK Interface
DEFAULT_SYSTEM_PROMPT = (
"You are nanobot, a concise, useful private AI agent. You can name instruments when "
"they assist. Prefer utilizing a device over guessing for math, the present time, operating "
"code, net lookups, or recalling saved details. After instruments run, reply the consumer "
"immediately and clearly."
)
class Nanobot:
def __init__(self, supplier: Provider, *, system_prompt: str = DEFAULT_SYSTEM_PROMPT,
token_budget: int = 3000, max_iterations: int = 6, verbose: bool = True):
self.registry = ToolRegistry()
self.reminiscence = Memory(token_budget=token_budget)
self.abilities: dict[str, Skill] = {}
self._loaded_skills: set[str] = set()
self._base_system = system_prompt
self.agent = Agent(supplier, self.registry, self.reminiscence,
system_prompt, max_iterations=max_iterations, verbose=verbose)
for t in (calculator, get_current_time, run_python, web_search):
self.registry.add(t)
@classmethod
def auto(cls, **kw) -> "Nanobot":
"""Pick a actual supplier if an API secret is set, else the Mock supplier."""
api_key = os.environ.get("NANOBOT_API_KEY") or os.environ.get("OPENAI_API_KEY")
mannequin = os.environ.get("NANOBOT_MODEL", "gpt-4o-mini")
base_url = os.environ.get("NANOBOT_BASE_URL")
if api_key and _HAVE_OPENAI:
print(f"→ Using reside supplier: OpenAI-compatible (mannequin={mannequin}, base_url={base_url or 'api.openai.com'})")
supplier: Provider = OpenAICompatibleProvider(api_key, mannequin, base_url)
else:
why = "no API key discovered" if not api_key else "openai SDK unavailable"
print(f"→ Using Mock supplier ({why}). Set NANOBOT_API_KEY for a reside mannequin.")
supplier = MockProvider()
return cls(supplier, **kw)
def add_tool(self, f: Callable) -> "Nanobot":
self.registry.add(device(f) if not isinstance(f, Tool) else f)
return self
def register_skill(self, ability: Skill) -> "Nanobot":
self.abilities[skill.name] = ability
return self
def load_skill(self, title: str) -> "Nanobot":
"""Activate a ability: append its directions and register its instruments."""
sk = self.abilities[name]
if title not in self._loaded_skills:
self.agent.system_prompt += f"nn## Skill: {sk.title}n{sk.directions}"
for t in sk.instruments:
self.registry.add(t)
self._loaded_skills.add(title)
print(f" · loaded ability '{title}' (+{len(sk.instruments)} device(s))")
return self
def connect_mcp(self, server: MCPServer) -> "Nanobot":
for t in mcp_tools(server):
self.registry.add(t)
print(f" · related MCP server '{server.title}' (+{len(server.list_tools())} device(s))")
return self
async def run(self, message: str, *, session_key: str = "sdk:default",
hooks: Optional[list[AgentHook]] = None) -> RunConsequence:
return await self.agent.run(message, session_key=session_key, hooks=hooks)
class AuditHook(AgentHook):
"""Print each device the mannequin decides to name."""
def __init__(self):
self.calls: checklist[str] = []
async def before_execute_tools(self, context: AgentHookContext) -> None:
for tc in context.tool_calls:
self.calls.append(tc.title)
print(f" [audit] {tc.title}({tc.arguments})")
class TimingHook(AgentHook):
"""Measure how lengthy every LLM iteration takes."""
def __init__(self):
self._t = 0.0
async def before_iteration(self, context: AgentHookContext) -> None:
self._t = time.perf_counter()
async def after_iteration(self, context: AgentHookContext) -> None:
ms = (time.perf_counter() - self._t) * 1000
print(f" [timing] iteration {context.iteration} took {ms:.1f} ms")
class CensorHook(AgentHook):
"""finalize_content runs as a pipeline — remodel the ultimate textual content."""
def finalize_content(self, context: AgentHookContext, content material: str) -> str:
return content material.change("secret", "***") if content material else content material
async def demo_basic(bot: Nanobot):
banner("DEMO 1 — Basic chat (no instruments wanted)")
r = await bot.run("Hello! Who are you?", session_key="demo-basic")
print("AGENT:", r.content material)
print(f"(iterations={r.iterations}, instruments={r.tools_used}, ~tokens={r.utilization.whole})")
async def demo_tool_calling(bot: Nanobot):
banner("DEMO 2 — Tool calling: math, time, and Python")
for q in ["What is 2 ** 10 + sqrt(144)?",
"What time is it in Tokyo?",
"Write Python to list the first 12 Fibonacci numbers."]:
print(f"nUSER: {q}")
r = await bot.run(q, session_key="demo-tools")
print("AGENT:", r.content material)
async def demo_multistep(bot: Nanobot):
banner("DEMO 3 — Multi-step loop with an audit hook")
audit = AuditHook()
q = "Calculate 15 * 23, and additionally inform me the present time in Asia/Kolkata."
print(f"USER: {q}")
r = await bot.run(q, session_key="demo-multistep", hooks=[audit])
print("AGENT:", r.content material)
print("Tools noticed by hook:", audit.calls)
async def demo_memory(bot: Nanobot):
banner("DEMO 4 — Session reminiscence (unbiased histories per session_key)")
await bot.run("My title is Ada and I like Python.", session_key="user-ada")
await bot.run("My title is Alan and I like Haskell.", session_key="user-alan")
r1 = await bot.run("What's my title and what do I like?", session_key="user-ada")
r2 = await bot.run("What's my title and what do I like?", session_key="user-alan")
print("ADA session →", r1.content material)
print("ALAN session →", r2.content material)
print("(Each session_key saved its personal dialog historical past — like nanobot.)")
async def demo_skills(bot: Nanobot):
banner("DEMO 5 — Skills: load a 'analysis' functionality on demand")
analysis = Skill(
title="analysis",
description="Web analysis workflow",
directions=("When researching, first search the net, then synthesize the "
"snippets into a brief, sourced abstract."),
instruments=[web_search],
)
bot.register_skill(analysis).load_skill("analysis")
r = await bot.run("Search for the newest on retrieval-augmented era and summarize.",
session_key="demo-skills")
print("AGENT:", r.content material)
async def demo_mcp(bot: Nanobot):
banner("DEMO 6 — MCP-style exterior device server")
server = MCPServer("climate")
server.register(
title="forecast",
description="Get a (stub) climate forecast for a metropolis.",
parameters={"kind": "object",
"properties": {"metropolis": {"kind": "string"}},
"required": ["city"]},
handler=lambda metropolis: f"Forecast for {metropolis}: 27°C, partly cloudy (stub MCP information).",
)
bot.connect_mcp(server)
print("Registered instruments now embrace:", [n for n in bot.registry.names() if "weather" in n])
t = bot.registry.get("weather__forecast")
print("Direct MCP device name →", await t(metropolis="Delhi"))
async def demo_streaming_and_finalize(bot: Nanobot):
banner("DEMO 7 — finalize_content pipeline + timing hook")
q = "Compute sqrt(2) to point out the mathematics device, then reply."
print(f"USER: {q}")
r = await bot.run(q, session_key="demo-hooks", hooks=[TimingHook(), CensorHook()])
print("AGENT:", r.content material)
async def demo_capstone(bot: Nanobot):
banner("DEMO 8 — Capstone: a private agent juggling instruments + reminiscence")
print("A brief multi-turn 'private assistant' dialog:n")
turns = [
"What's 144 / 12, and what's my favorite language?",
"Run Python to print all primes under 50.",
]
for q in turns:
print(f"USER: {q}")
r = await bot.run(q, session_key="capstone", hooks=[AuditHook()])
print("AGENT:", r.content material, "n")
We wrap the lower-level agent in a Nanobot-style interface that feels extra like a actual SDK. We add help for registering instruments, loading abilities, connecting MCP-style servers, and operating the bot with session-specific reminiscence. We additionally outline a number of demo features that present primary chat, device calling, multi-step execution, reminiscence, abilities, MCP instruments, and hooks in motion.
Adding Long-Term Memory and Running the Demos
_FACTS: dict[str, str] = {}
@device
def remember_fact(key: str, worth: str) -> str:
"""Store a reality in long-term key-value reminiscence.
key: brief identifier
worth: the worth to retailer"""
_FACTS[key] = worth
return f"Stored {key} = {worth}"
@device
def recall_fact(key: str) -> str:
"""Recall a beforehand saved reality by key.
key: the identifier used when storing"""
return _FACTS.get(key, f"(no reality saved beneath '{key}')")
async def principal():
banner("
nanobot-from-scratch — constructing & operating the core structure")
bot = Nanobot.auto(verbose=True)
bot.add_tool(remember_fact).add_tool(recall_fact)
print("Registered instruments:", bot.registry.names())
await demo_basic(bot)
await demo_tool_calling(bot)
await demo_multistep(bot)
await demo_memory(bot)
await demo_skills(bot)
await demo_mcp(bot)
await demo_streaming_and_finalize(bot)
await demo_capstone(bot)
banner("DONE")
print(textwrap.dedent("""
You simply constructed nanobot's core: a provider-agnostic agent loop with instruments,
token-budgeted session reminiscence, lifecycle hooks, abilities, and an MCP-style device
server — the identical structure HKUDS/nanobot ships, saved intentionally small.
── Run the REAL nanobot ─────────────────────────────────────────────────────
!pip set up nanobot-ai
# configure a supplier + mannequin in ~/.nanobot/config.json, then:
from nanobot import Nanobot as ActualNanobot
bot = ActualNanobot.from_config()
consequence = await bot.run("What time is it in Tokyo?")
print(consequence.content material)
Docs: https://github.com/HKUDS/nanobot • Python SDK: docs/python-sdk.md
"""))
def _go():
attempt:
asyncio.run(principal())
besides RuntimeError:
loop = asyncio.get_event_loop()
loop.run_until_complete(principal())
if __name__ == "__main__":
_go()
We add easy long-term key-value reminiscence instruments to retailer and recall details. We outline the primary execution operate that creates the bot, registers customized instruments, and runs each demo from begin to end. We full the tutorial by exhibiting how the rebuilt nanobot-style structure connects to the true nanobot bundle for future extension.
Conclusion
In conclusion, we have now a working nanobot-style agent that may name instruments, retain session-specific context, load abilities, hook up with exterior device servers, and run a clear, provider-agnostic loop. We additionally perceive how a small and readable structure can help highly effective agent habits with out counting on a heavy orchestration layer. It provides us leverage to increase the agent additional with actual LLM suppliers, manufacturing instruments, persistent reminiscence, and customized abilities for extra superior private AI workflows.
Check out the Full Codes here. Also, be happy to observe us on Twitter and don’t overlook to hitch our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to accomplice with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and so on.? Connect with us
The submit Build a Nanobot-Style AI Agent in Google Colab with Tool Calling, Session Memory, Skills, and MCP Servers appeared first on MarkTechPost.
