Everyone I know learning agentic AI right now is learning to use frameworks. They call .invoke(), something happens, they move on.
Ask them what fires when a tool class is defined. Ask them why validation should live in a descriptor instead of __init__. Ask how ParamSpec preserves type safety through decorator stacks. Blank stares.
I wanted to actually understand this stuff. So I built a mini Python framework that mirrors the internal shape of real agent systems — registry, validated config, mixins, decorators, sessions, streaming, TypedDicts, Protocols, and a CLI. Zero external dependencies. The whole thing is explainable from source.
Here's the full breakdown.
Architecture at a glance
agent_cli/
├── core/
│ ├── base.py ← BaseTool: __init_subclass__, run(), stream()
│ ├── registry.py ← ToolRegistry: global catalog of tool classes
│ ├── config.py ← ToolConfig: descriptor-validated settings
│ ├── mixins.py ← LoggingMixin, RetryMixin, MetricsMixin
│ ├── session.py ← ExecutionSession: context manager lifecycle
│ ├── metrics.py ← ToolMetrics: slotted execution counters
│ ├── protocols.py ← ToolProtocol: structural typing contract
│ ├── types.py ← ToolContext, ToolOutput, ToolMetadata TypedDicts
│ └── exceptions.py ← AgentCliError hierarchy
├── tools/
│ ├── search.py ← SearchTool
│ ├── summarize.py ← SummarizeTool
│ └── translate.py ← TranslateTool
├── decorators/
│ ├── tooling.py ← @tool (discovery metadata)
│ └── execution.py ← @log_execution, @measure_time (ParamSpec)
├── descriptors/
│ └── fields.py ← ValidatedField, IdentifierField, IntegerRange, FloatRange, BooleanField
├── cli/
│ └── app.py ← argparse CLI: list-tools, describe, run
└── tests/
└── test_framework.py
Let's go piece by piece.
1. Self-registering tools with __init_subclass__
The first real design question in any agent framework: how does the runtime know what tools exist?
The naive approach is a dict you maintain by hand. The problem is obvious — add a tool class, forget the dict entry, get a silent "tool not found" error later. It's a real, common bug.
The fix: __init_subclass__ on the base class. It fires at class definition time, not instantiation.
class BaseTool(LoggingMixin, RetryMixin, MetricsMixin, ABC):
def __init_subclass__(
cls,
*,
tool_name: str | None = None,
description: str = "",
streamable: bool = False,
abstract: bool = False,
**kwargs: Any,
) -> None:
super().__init_subclass__(**kwargs)
if abstract:
return
if tool_name is None:
raise TypeError(f"{cls.__name__} must define tool_name='...'")
cls._tool_name = tool_name.strip().lower()
cls.description = description.strip()
cls._streamable = streamable
ToolRegistry.register(cls._tool_name, cls)
A tool just declares itself:
class SearchTool(BaseTool, tool_name="search", streamable=True):
def execute(self, context: ToolContext) -> str:
...
The moment Python parses that class — SearchTool is in the registry. No manual list. Forget tool_name= and you get TypeError immediately, not a confusing failure later.
The ToolRegistry stores a flat dict[str, type] and raises DuplicateToolError if two classes try to claim the same name:
@classmethod
def register(cls, name: str, tool_cls: type[Any]) -> None:
existing = cls._tools.get(name)
if existing is not None and existing is not tool_cls:
raise DuplicateToolError(
f"tool name {name!r} is already registered by {existing.__name__}"
)
cls._tools[name] = tool_cls
The test proves registration happens purely from import:
def test_builtin_tools_register_automatically(self) -> None:
# import agent_cli.tools triggers __init_subclass__ on all three
self.assertEqual(("search", "summarize", "translate"), ToolRegistry.names())
2. Validated config with data descriptors
Bad config values that propagate silently into retry loops are painful to debug. The fix is to validate at assignment time, not at execution time.
Python data descriptors intercept __set__:
class ValidatedField(Generic[T]):
def __set_name__(self, owner, name):
self.public_name = name
self.private_name = f"_{name}"
def __set__(self, instance, value):
setattr(instance, self.private_name, self.validate(value))
def validate(self, value):
return value # override in subclasses
Concrete examples:
class IdentifierField(NonEmptyString):
_pattern = re.compile(r"^[a-z][a-z0-9_-]*$")
def validate(self, value: Any) -> str:
cleaned = super().validate(value)
if not self._pattern.fullmatch(cleaned):
raise ValueError(f"{self.public_name} must be a valid identifier")
return cleaned
class IntegerRange(ValidatedField[int]):
def validate(self, value: Any) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise TypeError(f"{self.public_name} must be an integer")
if not self.min_value <= value <= self.max_value:
raise ValueError(f"{self.public_name} must be between {self.min_value} and {self.max_value}")
return value
ToolConfig uses them as class-level declarations:
class ToolConfig:
__slots__ = ("_retries", "_streaming_enabled", "_timeout", "_tool_name")
tool_name: str = IdentifierField()
retries: int = IntegerRange(0, 5, default=1)
timeout: float = FloatRange(0.1, 120.0, default=10.0)
streaming_enabled: bool = BooleanField(default=True)
@property
def reliability_profile(self) -> str:
if self.retries >= 3 and self.timeout >= 15:
return "resilient"
if self.retries == 0:
return "fast-fail"
return "balanced"
ToolConfig(tool_name="Bad Name", retries=9) raises two ValueErrors at construction. The tests cover this explicitly:
def test_invalid_config_fails_fast(self) -> None:
with self.assertRaises(ValueError):
ToolConfig(tool_name="Bad Name", retries=1, timeout=10.0)
with self.assertRaises(ValueError):
ToolConfig(tool_name="search", retries=9, timeout=10.0)
3. Mixins composing via MRO
class BaseTool(LoggingMixin, RetryMixin, MetricsMixin, ABC):
Three mixins, each owning one concern. All __slots__ = () — no instance state, pure behavior:
LoggingMixin — scoped self.log(message) that prefixes the tool name.
RetryMixin — self.with_retries(operation) wraps any callable:
def with_retries(self, operation: Callable[[], R]) -> R:
last_error = None
for attempt in range(1, self.config.max_attempts + 1):
try:
return operation()
except Exception as error:
last_error = error
FrameworkLogger.warning(
f"{self.config.tool_name}: attempt {attempt}/{self.config.max_attempts} failed: {error}"
)
raise last_error
MetricsMixin — self.record_metric() and self.metrics, backed by a slotted ToolMetrics that tracks runs, failures, and total duration.
You can inspect the full chain at runtime:
$ python main.py describe search
MRO: SearchTool -> BaseTool -> LoggingMixin -> RetryMixin -> MetricsMixin -> ABC -> object
4. ParamSpec decorators that preserve type signatures
Naive decorators erase type signatures. mypy sees Callable[..., Any] instead of the real thing. ParamSpec fixes this:
P = ParamSpec("P")
R = TypeVar("R")
def log_execution(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
owner = args[0].__class__.__name__ if args else func.__module__
FrameworkLogger.info(f"starting {owner}.{func.__name__}")
try:
result = func(*args, **kwargs)
except Exception as error:
FrameworkLogger.error(f"failed {owner}.{func.__name__}: {error}")
raise
FrameworkLogger.info(f"finished {owner}.{func.__name__}")
return result
return wrapper
BaseTool.run() stacks both decorators:
@log_execution
@measure_time
def run(self, raw_input: str, *, stream: bool = False, session_id: str = "standalone") -> ToolOutput:
...
mypy still sees the full run() signature after both wrappers. This is the pattern production frameworks use when they layer instrumentation on top of user-facing methods.
5. ExecutionSession: context manager lifecycle
Every tool run happens inside a session:
with ExecutionSession() as session:
session.add_resource(f"{tool.name}-runtime")
result = tool.run(raw_input, stream=args.stream, session_id=session.session_id)
__enter__ records the start time and logs session start. __exit__ calls cleanup() in all cases — success or exception — releases tracked resources in reverse order, and logs total duration. __exit__ returns False so exceptions propagate normally.
Each session generates a unique session_id (session-{uuid4().hex[:10]}) that flows into ToolContext and out through ToolOutput, so every result is traceable.
6. Generator streaming + TypedDict contracts
Both execution paths produce the same ToolOutput:
class ToolOutput(TypedDict):
tool: str
content: str
tokens: list[str]
duration_ms: float
session_id: str
Streaming uses a real generator:
def stream(self, context: ToolContext) -> Iterator[str]:
for token in self.execute(context).split(" "):
yield f"{token} "
BaseTool.run() collects generator output and records metrics in finally regardless of path:
try:
if stream:
tokens = self.with_retries(lambda: list(self.stream(context)))
content = "".join(tokens).strip()
else:
content = self.with_retries(lambda: self.execute(context))
tokens = self._tokenize(content)
return {"tool": self.name, "content": content, "tokens": tokens, ...}
except Exception:
failed = True
raise
finally:
self.record_metric(duration_ms=..., failed=failed)
The finally block guarantees metrics are always recorded — including on failure. That's important for production observability.
7. Structural typing with Protocol
The registry returns ToolProtocol, not BaseTool:
@runtime_checkable
class ToolProtocol(Protocol):
@property
def name(self) -> str: ...
@property
def metadata(self) -> ToolMetadata: ...
def execute(self, context: ToolContext) -> str: ...
def stream(self, context: ToolContext) -> Iterator[str]: ...
def run(self, raw_input: str, *, stream: bool, session_id: str) -> ToolOutput: ...
The CLI depends on the contract, not the class. A third-party tool that implements ToolProtocol without inheriting BaseTool is fully compatible. This is exactly how production frameworks stay open for extension.
This is Week 1 of a 16-week public build series. Python → NLP → embeddings → LLM internals → RAG → multi-agent → MCP → deployment. Each week gets a build and a post, extending this same codebase.
SOCIAL SHARE CARD GENERATOR