DEV Community: Sajid Islam

I built a mini agent framework in Python to understand how LangGraph actually works under the hood

Sajid Islam — Tue, 23 Jun 2026 13:48:19 +0000

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.

Repo: https://github.com/Sajid0875/agentic-systems-bootcamp/tree/main/Week-01-Python-Agentic-Systems/Session%201/project_agent_cli%20

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.

I built a mini agent-tool framework to actually understand how LangGraph/CrewAI register tools

Sajid Islam — Sun, 21 Jun 2026 10:52:00 +0000

I'm going through a 16-week Agentic AI syllabus right now, and Week 1 is "Python for Agentic Systems" — OOP, typing, decorators. Instead of just reading about it, I built a small CLI toolkit that mimics how real agent frameworks register and run tools.

This post is about one piece of it: how a global tool registry works using __init_subclass__, and why agent frameworks need this pattern at all.

Repo's at the bottom. Code below is real, from the actual project — not pseudocode.

The problem

Agent frameworks (LangGraph, CrewAI, PydanticAI) all need the same thing: a way for an LLM or planner to discover "what tools exist" without you manually maintaining a list somewhere.

The naive way is a dict you update by hand:

TOOLS = {
    "search": SearchTool,
    "summarize": SummarizeTool,
}

This works until you forget to add an entry. Then your planner silently can't find a tool that exists in your codebase.

The fix: self-registering classes

Here's the actual registry from my project:

class ToolRegistry:
    __slots__ = ()
    _tools: dict[str, type[Any]] = {}

    @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

And the part that actually calls register() — __init_subclass__ on the base class:

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)

__init_subclass__ fires automatically the moment Python defines a subclass — before you ever instantiate it. So a tool just declares itself:

class SearchTool(
    BaseTool,
    tool_name="search",
    description="Searches a small in-memory knowledge base.",
    streamable=True,
):
    def execute(self, context: ToolContext) -> str:
        ...

The moment this class body is parsed, SearchTool is in the registry. No manual list. No import-time side-effect hacks. Forget tool_name= and you get a TypeError immediately — not a silent miss three files away.

Why this matters for agent frameworks specifically

Once tools self-register, a CLI (or a planner LLM) can just ask "what do you have":

def _list_tools() -> None:
    for name, tool_cls in ToolRegistry.items():
        tool = tool_cls()
        print(f"{name:<12} {tool.metadata['description']}")

$ python main.py list-tools
search       Searches a small in-memory knowledge base and returns ranked notes.
summarize    Creates a compact extractive summary of user-provided text.
translate    Translates common demo phrases to Spanish or Urdu using a local lexicon.

This is structurally the same problem LangGraph and CrewAI solve with their own tool-discovery mechanisms. Different implementation, same underlying need: a single source of truth that updates itself.

What's also in the project

This registry is one piece. The same codebase has:

Descriptors (ValidatedField, IdentifierField, IntegerRange) validating ToolConfig at assignment time
Mixins + MRO — BaseTool(LoggingMixin, RetryMixin, MetricsMixin, ABC) composes logging, retries, and metrics without inheritance spaghetti
ParamSpec-based decorators (@log_execution, @measure_time) that wrap methods without breaking their signatures for mypy
Generator-based streaming — stream() yields tokens instead of faking it with string slicing

I'll cover each of these in upcoming posts as I move through the syllabus.

Try it

git clone https://github.com/Sajid0875/agentic-systems-bootcamp
cd agent-ready-cli-toolkit
python main.py list-tools
python main.py describe search
python main.py run summarize "Agent frameworks register tools and stream results." --stream

Repo: https://github.com/Sajid0875/agentic-systems-bootcamp/tree/main/Week%201%20/Session%201/project_agent_cli%20

If you're also learning agentic systems and want to compare notes on how different frameworks (CrewAI, PydanticAI, LangGraph) handle tool registration internally, drop it in the comments — genuinely curious how close/far off this mental model is from the real implementations.

Bridging the Gap: From Classical Search Theory to the Era of Agentic AI

Sajid Islam — Sat, 07 Mar 2026 13:50:12 +0000

Abstract

This paper is all about the evolution of Artificial Intelligence from foundational problem-solving agents to modern agentic systems. By synthesizing basic search algorithms like A*, Breadth-First Search (BFS), and Iterative Deepening with recent milestones in Large Language Model (LLM)-based search agents, we demonstrate how classical state-space theory provides the essential framework for autonomous intelligence. In this blog, we will analyze the shift from reactive generative models to proactive agents capable of dynamic planning, reflection, and multi-turn reasoning.

1. Introduction: The Agentic Evolution
In classical AI theory, a problem-solving agent is defined as one that plans ahead by finding sequences of actions that lead to favorable goals. Traditionally, these agents operated in static, fully observable, and deterministic environments using atomic representations. However, the industry is moving into "The Rise of Agentic AI," a shift toward systems that do not merely answer prompts but autonomously set sub-goals, select tools, and execute multi-step actions with minimal human supervision. This transition moves AI from a reactive assistant to a proactive collaborator.

2. Foundational Frameworks: State Space and Representations

Modern agents navigate environments of immense complexity, yet they rely on the same representational structures taught in foundational search theory:
Atomic vs. Structured States: While early search problems used atomic representations (states as "black boxes"), modern agentic AI utilizes Structured Representations. This allows agents to understand relationships between objects, which is critical for navigating the World Wide Web as a state space.
The Transition Model: In theory lectures, the RESULT(s, a) function gives us the outcome of an input (action). In agentic systems, this is scaled into World Modeling, where the agent senses its environment to decide on the next optimal action in their environment.

3. Scaling Search: From Algorithms to Agents

The search strategies analyzed in the classroom—such as Uninformed Search (BFS, DFS) and Informed Search (A*)—are the basics of modern Deep Search Agents.
3.1. Test-Time Scaling and Tree Search

Classical algorithms like Iterative Deepening Search (IDS) save memory while exploring deep trees. Modern "Deep Research" systems use a similar philosophy through Test-Time Scaling. By allocating more computation during inference, these agents use techniques like Monte Carlo Tree Search (MCTS) and Self-Consistency to explore multiple reasoning paths before committing to an answer.

3.2. Heuristics and Reward Functions

In informed search, the Heuristic Function estimates the cost from the current state to the goal. In the world of agentic AI, this is mirrored by Multi-objective Reward Functions. Agents use Reinforcement Learning (RL) to "calculate" the relevance and cost of information retrieval, effectively treating web navigation as an f(n) = g(n) + h(n) optimization problem.

4. Architectural Paradigms

Research identifies several core architectures that enable agency:
BDI (Belief-Desire-Intention): A cognitive framework where the agent acts based on its "Beliefs" (knowledge), "Desires" (goals), and "Intentions" (commitments).
ReAct (Reasoning + Acting): A loop where the agent "Thinks" to generate a plan and "Acts" to execute it, observing the result to refine the next step.
Hierarchical Models: A supervisor agent delegates complex tasks to specialized worker agents, mirroring the way complex search problems are decomposed into sub-problems.

5. Challenges in the Real World

Despite their potential, agentic systems face significant hurdles that go beyond classroom simulations:
Cascading Errors: Unlike a simple route-finding problem in Romania, a small reasoning error in an autonomous agent can propagate through a "long planning chain," leading to total failure.
Reliability and Hallucination: Agents may "hallucinate" or create false data during multi-turn retrieval.
Governance: As agents become more autonomous, determining accountability for their decisions remains a critical ethical challenge.

6. Conclusion

The evolution from Uninformed Search to Autonomous Agentic AI represents one of the most significant changes in computer science. By grounding modern LLM-based search agents in the formal definitions of state spaces, transition models, and heuristics, we can build systems that are not just smarter at talking, but smarter at acting. As the market for these agents is projected to exceed $50 billion by 2030, the synthesis of theory and practice remains the most vital for AI researchers.

Summary & Connection:

The Transition to Agentic Search
The primary goal of the research papers is to define and categorize the shift from Generative AI (which focuses on pattern-based content creation) to Agentic AI and Deep Search Agents, which are autonomous, goal-driven systems capable of multi-step reasoning, planning, and environmental interaction. These agents move beyond static, single-turn responses to execute multi-turn dynamic retrieval across diverse information sources like the web and private databases.
This research directly evolves the foundational logic of Problem-Solving Agents introduced in our course material. While our lectures define an agent’s operation through a four-phase process (Goal formulation, Problem formulation, Search, and Execution), modern agentic AI enhances this cycle with Memory (long-term and short-term) and Reflection/Evaluation modules that allow the agent to self-critique and refine its trajectory.
Specifically, these findings modify and scale our course learning in three ways:
Search Structures: While we studied uninformed search (BFS, DFS) and informed search (A*) as linear or simple tree paths, modern agents utilize Hybrid Structures, such as Monte Carlo Tree Search (MCTS) and Graph-based search, to backtrack and dynamically revise decisions.
Heuristics and A*: In class, Informed Search relies on a heuristic function h(n) to estimate the cost to the goal. In Agentic AI, this is replaced by Multi-objective Reward Functions in Reinforcement Learning, which calculate the "cost" and "gain" of information retrieval based on efficiency, diversity, and evidence quality.
State Representations: The "State Space" discussed in our lectures (atomic, factored, structured) is scaled to handle the entire internet, where modern agents must use Structured Representations to normalize raw, multimodal data into internal models for autonomous planning.

Personal Insight:

Manual Reading vs. NotebookLM Exploration
My manual reading of the papers allowed me to grasp the high-level shift toward "agenticness" and the conceptual differences between frameworks like ReAct (interleaving reasoning and acting) and the BDI (Belief-Desire-Intention)architecture. I understood that "Deep Search" is effectively a way to scale up "test-time search", meaning the more an agent is allowed to search and "think" during inference, the more accurate it becomes.
However, my NotebookLM exploration revealed a deeper technical synthesis that was not immediately obvious:

*Heuristics as Reward Models: * NotebookLM helped me realize that the Action Cost Functions we studied in route-finding problems are the direct ancestors of the Reinforcement Learning rewards used to train agents like OpenAI’s Deep Research.

The Frontier of Iterative Deepening: It clarified that modern "Reflection-Driven Sequential Search" is essentially a high-level, dynamic implementation of Iterative Deepening Search (IDS), where the agent repeatedly applies reasoning with increasing complexity until it satisfies a "Goal Test".

Test-Time Scaling vs. Complexity: I found it fascinating that while we learn that traditional search is limited by exponential space complexity (O(b^d)), modern agents use Test-Time Scaling to prioritize "computational latency" as the primary cost, trading time for accuracy in a way that parallels the Weighted A* strategies from our lecture.

@raqeeb_26