DEV Community: Shubham Srivastava

Episode 1 : The Beginning of the Agentic Era

Shubham Srivastava — Thu, 09 Jul 2026 11:47:01 +0000

PREFACE

Large Language Models have transformed how we build software, but most applications barely scratch the surface of their capabilities. The real shift isn't just using an LLM—it's building intelligent agents that can reason, use tools, maintain context, and solve complex tasks. In this series, we'll explore the foundations of the agentic era and learn how to engineer systems that truly scale.

Today, we have numerous agentic frameworks that make it easy to build anything from simple AI agents to sophisticated multi-agent systems powered by RAG, MCP, and tool calling. While these abstractions are incredibly useful, they often hide the core principles, design decisions, and execution flow that make agentic systems work.

In this series, we'll peel back those abstractions and build our own agentic framework from scratch. The goal isn't to replace existing frameworks—it's to understand the architecture, reasoning, orchestration, and every moving part that powers modern AI agents. By the end, you'll know not just how to use these frameworks, but why they work the way they do.

Reality and Hype

Large Language Models are incredibly powerful—they can generate content, reason over information, use tools, and enable autonomous workflows. However, they are still non-deterministic systems, meaning they cannot guarantee 100% accurate or consistent outputs. Their responses are probabilistic by nature.

This is why not every software problem needs an LLM, an AI assistant, or an agent. Before introducing AI into a system, ask a simple question: Does this problem require reasoning, handling ambiguity, or making decisions in situations where the outcome isn't strictly predefined? If the answer is yes, an LLM—combined with proper guardrails, governance, observability, and human oversight—can be a great fit. If the problem is deterministic and follows well-defined business rules, a traditional software solution will almost always be simpler, faster, cheaper, and more reliable.

The goal isn't to add AI everywhere—it's to use it where it creates real value.

1.0 The 10,000-Foot View of Agentic AI

Before we dive into building agentic systems, it's important to understand one fundamental idea: everything revolves around the LLM. It acts as the reasoning engine that interprets instructions, decides what to do next, and generates responses.

However, LLMs have two important characteristics. First, they are probabilistic, meaning they generate responses based on probabilities rather than deterministic rules, so the same input may not always produce the exact same output. Second, they are stateless—they don't remember previous conversations or actions on their own. Every request is independent, which means we are responsible for maintaining the conversation history, context, and any relevant state, and sending it back to the LLM with each request.

You can think of an LLM as the brain or boss of an agentic system. It doesn't execute code, call APIs, or access databases directly. Instead, it understands the context you provide, reasons about the problem, and outputs instructions in text (or structured tool calls). The surrounding application is responsible for carrying out those instructions and feeding the results back to the LLM.

1.1 Tool Calling

As we've discussed, an LLM doesn't execute code, query databases, or call external APIs on its own—and that's by design. Its primary responsibility is reasoning and generating the next best action based on the context it receives. Offloading execution to external tools allows the LLM to focus on what it does best: understanding language, reasoning over information, and making decisions.

One of the most powerful capabilities of modern LLMs is tool calling. Instead of performing an action directly, the model can decide which tool to use and generate a structured tool call containing the tool name and the required arguments. Your application then executes the tool and returns the result back to the LLM for further reasoning.

It's worth noting that not every LLM supports tool calling. While most modern frontier models from providers such as OpenAI, Anthropic, and Google do, some lightweight or smaller reasoning models may not. Always verify that your chosen model supports tool calling before building an agentic workflow.

Under the hood, models that support tool calling are trained to generate structured tool calls through techniques such as Supervised Fine-Tuning (SFT), enabling them to reliably produce JSON-like outputs that conform to predefined tool schemas.

For example, when using the OpenAI API, tools are defined using a JSON schema like the one below:

# 1. Define a list of callable tools for the model
tools = [
    {
        "type": "function",
        "name": "get_horoscope",
        "description": "Get today's horoscope for an astrological sign.",
        "parameters": {
            "type": "object",
            "properties": {
                "sign": {
                    "type": "string",
                    "description": "An astrological sign like Taurus or Aquarius",
                },
            },
            "required": ["sign"],
        },
    },
]

Once you've created the OpenAI client, you can pass the list of tool definitions along with your request. This makes the model aware of the tools it is allowed to use. Whenever the LLM determines that a user's request requires one of those tools, it doesn't execute the tool itself. Instead, it generates a tool call containing the tool name and the required arguments in a structured format. Your application is then responsible for executing the tool and sending the result back to the LLM so it can continue reasoning and generate the final response.

# 2. Prompt the model with tools defined
response = client.responses.create(
    model="gpt-5.5",
    tools=tools,
    input=input_list,
)

1.2 Memory Management

Since LLM's are stateless(It's Ghajini😁(Indian Movie where hero need external source to remember)). Generally LLM's are trained to consume long term conversation in terms of roles that what they fine tuned for to better understand the context. So we can store this conversation while while keeping track of the roles for each conversation.

For our framework, we'll introduce the concept of sessions, where each session maintains its own conversation history and context. To keep things simple while still demonstrating persistence, we'll store each session on the local file system. This allows conversations to survive application restarts, and we'll also implement session lifecycle management, including creating, updating, and deleting session data when it's no longer needed.

1.3 Context Management

In an agentic framework,

💾 Memory Management and 🎛️ Context Management handle two entirely different stages of the data lifecycle.

💾 Memory Management is the Storage and Retrieval Layer. It dictates how an agent's experiences, facts, and trajectories are persisted over time. This component manages database writes to systems like Redis for short-term history, or Vector Databases for long-term semantic knowledge. Its primary job is data preservation and indexing.

🎛️ Context Management is the Runtime Governance Layer. It sits between your memory stores and the LLM. Because context windows (e.g.,128k tokens) are bounded and expensive, this layer dynamically selects, trims, and formats the retrieved memory data into the final prompt payload. Its primary job is token optimization and relevance.

Handling Long Contexts

One of the biggest challenges in building an agentic AI framework is managing the context window of the LLM. As agents interact with users, call tools, and reason over multiple steps, the conversation history continues to grow. Since every model has a finite context window, sending the entire history with every request eventually becomes impossible.

A common solution is context compression. Instead of retaining every message verbatim, we summarize older parts of the conversation while preserving the essential information needed for future reasoning. Fortunately, LLMs excel at summarizing large amounts of text, making them well-suited for compressing conversation history without losing the most important context. This allows long-running agentic workflows to remain efficient while staying within the model's context limits.

The decision to summarize the conversation is handled by the any method such as is_summarization_needed() method. This method periodically checkpoints the current session while also monitoring the total number of tokens consumed, using the model's usage information. Once the token count approaches or exceeds a predefined threshold of the model's context window, the engine automatically triggers a summarization process. The summarized conversation then replaces the older message history, allowing the agent to preserve the most important context while staying within the model's context limits for future interactions.

1.4 The Orchestrator

Agents are the heart of our framework. From the moment an agent is initialized with an LLM, it becomes the central orchestrator responsible for coordinating the entire workflow. It manages conversation context, memory, session persistence, tool execution, interactions with the LLM, response handling, retries, error handling, security guardrails, observability, callbacks, and the overall execution lifecycle.

Below is a glimpse of how an agent will look in our framework. Rather than building a single agent for every use case, we'll design specialized agents optimized for different execution patterns.

MonoQAgent — A single-agent runtime designed for linear and deterministic workflows. The LLM reasons about the user's request, decides which tools to invoke, processes the results, and returns the final response. This is ideal for tasks such as "What's the weather in London?", "Summarize this YouTube video", or "Analyze this document".
SwarmAgent — A multi-agent orchestration framework where multiple specialized agents collaborate to solve complex problems. Each agent has a well-defined responsibility—for example, planning, research, coding, reviewing, or validation—and they communicate to produce a final result. This architecture is well suited for long-running, multi-step workflows that require collaboration and task decomposition.

As the series progresses, we'll continue extending the framework with more execution models, giving us the flexibility to build everything from simple AI assistants to production-grade autonomous agent systems.

2.0 Understanding Our First Agent called MonoAgent

Let's build our first agent, MonoAgent. This agent will be responsible for managing the complete execution lifecycle of a single-agent workflow, including tool calling, maintaining conversation history, handling different message roles, interacting with the LLM, and orchestrating the overall reasoning process.

Before we implement the agent itself, let's first understand how tools are defined and registered in our framework. A flexible tool system is essential, as it allows end users to easily create, register, and expose custom tools that the agent can discover and invoke during execution.

2.1 How we wire tools into our agent

There are 3 things you need to know :

ToolRegistry
ToolRegistryExecutor
FalconAukTool

The ToolRegistry acts as the central repository for all registered tools. Internally, it maintains a key-value mapping where the key is the tool's unique name and the value is the corresponding tool instance.

The ToolRegistryExecutor is responsible for registering tools into a specific ToolRegistry. It exposes a decorator that, when applied to a function, converts it into a framework-compatible tool and automatically registers it with the associated registry. This provides a clean and intuitive API for end users to define and register custom tools.

Every registered tool is represented internally as a FalconAukTool. This abstraction exposes a to_model_specific(model_provider: ModelProvider) method, which converts the generic tool definition into the provider-specific schema required by the target LLM. Since providers such as OpenAI, Anthropic, and Gemini each expect different tool definitions and request formats, this abstraction allows our framework to support multiple model providers without changing the user-defined tool.

This design also enables tool isolation. Each agent can be initialized with its own ToolRegistry, giving it access only to the tools it requires. For example, a MonoAgent may be configured with one set of tools, while a SwarmAgent can use an entirely different registry. This separation improves modularity, security, and maintainability by ensuring that agents only have access to the capabilities they are explicitly granted.

class ToolRegistry:
    def __init__(self):
        self._registry: Dict[str, Tool] = {}
        self.current_model_provider: str = (
            None  # This will hold the current model provider
        )

    def register_tool(self, tool_instance: Tool):
        self._registry[tool_instance.name] = tool_instance

    def get_tool(self, name: str) -> Optional[Tool]:
        return self._registry.get(name)

    def get_all_schemas(self, target_provider: ModelProvider) -> List[Dict]:
        return [t.to_model_specific(target_provider) for t in self._registry.values()]

Now we can see the ToolRegistryExecutor which takes the ToolRegistry instance and registers all the tools created.

class ToolRegistryExecutor:
    def __init__(self, context: ToolRegistry):
        self.context = context

    def tool(self, name: str, description: str):
        def decorator(func):
            tool = FalconAukTool(
                name=name,
                description=description,
                func=func,
            )

            self.context.register_tool(tool)
            return tool

        return decorator

Inside ToolRegistryExecutor, the tool method acts as the decorator used to define and register tools. When a function is decorated, it is automatically wrapped by FalconAukTool, which serves as the framework's internal abstraction for all tools.

Rather than exposing the raw Python function directly, FalconAukTool encapsulates the tool's metadata, schema, and execution logic. It also provides the to_model_specific(model_provider: ModelProvider) method, which converts the generic tool definition into the provider-specific format required by the target LLM. Since providers such as OpenAI, Anthropic, and Gemini each define tool schemas differently, this abstraction allows the same tool implementation to work seamlessly across multiple model providers without requiring any changes from the developer.

def to_model_specific(self, model_provider: ModelProvider):

        match model_provider:
            case ModelProvider.OPENAI:
                return self._to_openai_tool()
            case ModelProvider.ANTHROPIC:
                raise NotImplementedError(
                    "Anthropic tool conversion is not implemented yet."
                )
            case ModelProvider.GEMINI:
                raise NotImplementedError(
                    "Gemini tool conversion is not implemented yet."
                )

Example for OpenAI it will look like :

def _to_openai_tool(self):
        _type_map = {
            str: "string",
            int: "integer",
            float: "number",
            bool: "boolean",
            list: "array",
            dict: "object",
            type(None): "null",
        }

        sig = inspect.signature(self.func)
        properties = {}
        required = []

        for name, param in sig.parameters.items():
            raw_type = (
                param.annotation if param.annotation != inspect.Parameter.empty else str
            )
            json_type = _type_map.get(raw_type, "string")

            properties[name] = {
                "type": json_type,
            }

            if param.default is inspect.Parameter.empty:
                required.append(name)

        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required,
                },
            },
        }

Now that we've explored how tool registration works internally, let's see how an end user can define and register custom tools for different agents. The framework is designed to keep the developer experience simple, allowing users to create tools with minimal boilerplate while the framework takes care of registration, metadata, and model-specific conversions behind the scenes.

First create instance for ToolRegistry and ToolRegistryExecutor

tool_registration = ToolRegistry()
registry = ToolRegistryExecutor(tool_registration)

Now create tool like :

@registry.tool(
    name="get_weather",
    description="Provides the current weather based on the location.",
)
def get_weather(location: str) -> str:
    return f"The current weather of the {location} is 21 degree celcius and rainy and cold."

Now we can use the ToolRegistry to retrieve all the registered tools, extract their metadata, and automatically convert them into the provider-specific schema required by the selected LLM. This abstraction allows the framework to generate the appropriate tool definitions for providers such as OpenAI, Anthropic, or Gemini without requiring any additional work from the developer.

For example, the following code generates the model-specific tool definitions from the registered tools:

tool_registration.get_all_schemas(ModelProvider.OPENAI)

This will give the response that we want to send to as tool_functions as required by the OpenAI. So, by calling above you will get an idea.

2.1 Under the Hood : Tool and Agent communication

First our agent provide the LLM with the list of tools that it have and says LLM.
🤖 Agent: "Here are the tools available to you. Use them whenever they're needed."

🧠 LLM: "Got it."

👤 User: "What's the weather in Paris?"

🧠 LLM: Thinking... 💭 "I don't know the current weather, but I have access to a get_weather(location) tool. I'll ask the agent to execute it with location = "Paris"."

🤖 Agent: "Received the tool request. Executing get_weather(location="Paris") and returning the result."

🧠 LLM: "Perfect! I now have the latest weather information. Using the tool's response, I can generate an accurate answer for the user."

🤖 Agent: "Final response delivered to the user."

2.2 Managing conversation History

As of now we are using the simple list to manage all these conversation hostory.

messages: List[BaseMessage] = [
            SystemMessage(content=self.system_prompt),
            UserMessage(content=user_input),
        ]

Once everything is set up, we start the agent by calling the run() method of our MonoAgent. This method orchestrates the entire execution loop, maintaining the conversation history by continuously appending AIMessage and ToolMessage objects as the LLM reasons, invokes tools, receives their results, and generates the final response.

# Tool Message Response
messages.append(
                        ToolMessage(
                            content=result,
                            tool_call_id=tc.id,
                            name=tc.function["name"],
                        )
                    )
# LLM Response
messages.append(response.message)

2.3 Managing different Responses from different Models

Since every model can send use different LLM response that we need to one single interface so that it can be utilized by our Agents for that we have single LLMResponse and adapter classes for each models.

class LLMResponse:
    def __init__(
        self,
        message: AssistantMessage,
        usage: Usage,
        raw_response: Any = None,
    ):
        self.message = message
        self.usage = usage
        self.raw_response = raw_response

You can find the complete code about this setup in the Github Repo.

2.3 Response from LLM Provider

First see how our provider looks like with OpenAIProvider.

class OpenAILLMProvider(BaseLLMProvider):
    def __init__(
        self,
        api_key: str,
        model: str = "gpt-4o",
        callbacks: Optional[CallbackManager] = None,
        **kwargs,
    ):
        super().__init__(api_key=api_key, model=model, callbacks=callbacks, **kwargs)
        self._base_url = kwargs.get("base_url")
        self._max_tokens = kwargs.get("max_tokens", 4096)
        self._adapter = OpenAIAdapter()

At this stage, you can initialize your LLM using either the provider's native API key (such as OpenAI or Anthropic) or a compatible provider like Groq by supplying the GROQ_API_KEY along with its base_url.

Now, let's look at the generate() method. Notice that by simply passing the tool registry, the LLM automatically receives the tool definitions in the schema expected by that specific model. This abstraction allows us to register tools once while letting the underlying provider handle the model-specific tool-calling format.

def generate(
        self,
        messages: list[BaseMessage],
        tool_registry: Optional[ToolRegistry] = None,
        **kwargs,
    ) -> LLMResponse:

        client = OpenAI(api_key=self._api_key, base_url=self._base_url)
        raw_messages = self._adapter.convert_messages(messages)
        raw_tools = (
            self._adapter.convert_tools(tool_registry)
            if tool_registry
            else None
        )

        params = {
            "model": kwargs.get("model", self._model),
            "messages": raw_messages,
            "max_tokens": kwargs.get("max_tokens", self._max_tokens),
        }
        if raw_tools:
            params["tools"] = raw_tools

        response = client.chat.completions.create(**params)
        normalized = self._adapter.normalize_response(response)

        return LLMResponse(
            message=AssistantMessage(
                content=normalized.content, tool_calls=normalized.tool_calls
            ),
            usage=normalized.usage,
            raw_response=response,
        )

This LLM provider will be shared across our MonoAgent and any future agents we build. Now, let's return to the MonoAgent and walk through its run() method, where all the pieces we've built so far come together to drive the complete agent execution flow.

def run(self, user_input: str, max_iters: int = 10, **kwargs) -> LLMResponse:
        tool_runner = ToolRunner(self.tools) if self.tools else None

        messages: List[BaseMessage] = [
            SystemMessage(content=self.system_prompt),
            UserMessage(content=user_input),
        ]
        for i in range(max_iters):
            response = self.provider.generate(
                messages=messages, tool_runtime_context=self.tools
            )

            messages.append(response.message)
            tokens = response.usage.total_tokens if response.usage else 0

            if not response.message.tool_calls:
                return response

            if tool_runner:
                for tc in response.message.tool_calls:
                    _cb(current_tool=tc.function["name"], tool_name=tc.function["name"])
                    _log(f"[MonoAgent] Executing tool: {tc.function['name']}")

                    result = tool_runner.execute(tc)
                    messages.append(
                        ToolMessage(
                            content=result,
                            tool_call_id=tc.id,
                            name=tc.function["name"],
                        )
                    )

        return response

The run() method is the heart of our MonoAgent. It implements the complete reason → act → observe loop that powers modern agentic systems.

We begin by creating a ToolRunner, which is responsible for executing any tools requested by the LLM. Next, we initialize the conversation history with a SystemMessage containing the agent's instructions and a UserMessage containing the user's query. This messages list becomes the single source of truth for the entire conversation and grows throughout the agent's execution.

The agent then enters an iterative loop, allowing it to reason and use tools multiple times before producing a final answer. During each iteration, we send the complete conversation history to the LLM by calling provider.generate(). Along with the messages, we also pass the available tools (tool runtime context), enabling the model to decide whether it should answer directly or request one or more tool calls.

Once the model responds, its AIMessage is appended to the conversation history. We also capture token usage, which can later be used for monitoring, analytics, or cost estimation.

At this point, the agent checks whether the model requested any tool calls. If no tools are required, the reasoning process is complete and the response is immediately returned to the user.

If the model does request tools, the ToolRunner executes each one. Before execution, we invoke callbacks and logging so the application can stream progress, display tool activity, or collect observability data. The tool is then executed with the arguments generated by the model, and its output is wrapped inside a ToolMessage.

Appending the ToolMessage back into the conversation is a crucial step. It gives the LLM access to the tool's result, allowing it to continue reasoning with fresh, real-world information instead of relying solely on its pretrained knowledge. The updated conversation history is then sent back to the model in the next iteration, repeating the cycle until no further tool calls are needed or the maximum number of iterations is reached.

This iterative reasoning loop is what transforms a plain LLM into an autonomous agent. Rather than producing a single response, the model can think, decide which tools to use, observe their outputs, and continue reasoning until it has enough information to generate a final answer.

2.4 Finally Using Our Agent

Now, let's see our agent in action and explore how an end user can interact with it. We'll walk through the complete flow—from receiving a user's request, reasoning about it, invoking tools when necessary, and finally generating the response.

We're also introducing a tool registration system that automatically registers tools for the type of agent you're building. For example, with a MonoAgent, you simply provide the ToolRegistry, and the agent takes care of tool discovery, schema generation, and execution behind the scenes. This keeps the API clean while hiding the underlying complexity.

tool_registration = ToolRegistry()
registry = ToolRegistryExecutor(tool_registration)

@registry.tool(
    name="get_weather",
    description="Provides the current weather based on the location.",
)
def get_weather(location: str) -> str:
    return f"The current weather of the {location} is 21 degree celcius and rainy and cold."

Now our agent :

mono_agent = MonoAgent(
    provider=OpenAILLMProvider(
        api_key="",
        model="openai/gpt-oss-20b",
        base_url="https://api.groq.com/openai/v1",
    ),
    tools=tool_registration,
    system_prompt="You are helpful assistant.",
    agent_callback=Listener(),
)

and final call run() method

response = mono_agent.run(user_input=user_input)

3.0 Summary

In this first episode, we laid the foundation for everything we'll build throughout this series. We explored the reality behind the hype surrounding Agentic AI, discussed when LLMs should—and shouldn't—be used, and developed a high-level understanding of how modern agents operate. We then broke down the interaction between the Agent, the LLM, and Tools, and implemented our own MonoAgent from scratch to understand the complete reasoning loop behind tool calling.

Along the way, we built a reusable LLM provider, designed the agent execution flow, managed conversation history using structured messages, and introduced an automatic tool registration system that keeps the developer experience simple while handling the complexity under the hood.

This is just the beginning. In the upcoming episodes, we'll continue evolving our framework by introducing memory, streaming, structured outputs, multi-agent communication, planning, observability, and many more production-grade capabilities—one concept at a time.

Thanks for reading! If you're interested in understanding Agentic AI from first principles instead of treating it as a black box, stay tuned for the next episode. The journey has only just begun. 🚀

Check out this Repo for complete and show love❤️ by liking the repository.

EPISODE 1 : JUST START IT TODAY MAKE IT BEAUTIFUL TOMORROW

Shubham Srivastava — Tue, 30 Jun 2026 04:52:19 +0000

Welcome! 👋

In this series, we're going to build a real-time backend system from scratch using a modern microservices architecture.

Our goal isn't just to build another project that works. Instead, we'll understand why we make certain architectural decisions, when to use specific technologies, and how individual services communicate to form a scalable distributed system.

We'll start by connecting all the pieces together. Along the way, we'll answer questions like:

What does it actually take to build a real-time system?
Which technologies are best suited for real-time communication, and why?
How do we design services that are independent, loosely coupled, and easy to scale?
How can we run multiple services locally without making development painful?
How do we containerize our applications with Docker so that moving from local development to deployment becomes seamless?
How do services communicate with each other in a distributed environment?
Where does an API Gateway fit into the architecture?
How do we secure our services?
What happens when things go wrong? How do we handle failures, retries, and unexpected errors gracefully?
What practices make a system reliable, maintainable, and production-ready?

Rather than jumping straight into coding, we'll first understand the architecture and the reasoning behind every component we introduce. Once we have a clear picture of the system, we'll implement each service step by step, gradually assembling a complete real-time backend.

By the end of this series, you won't just have a working project—you'll understand how modern backend systems are designed, how independent services collaborate, and how to build applications that are scalable, resilient, and ready for production.

So, let's get started and build something amazing together. 🚀

About Project

This project was built to understand how independent microservices run, communicate, and collaborate within a distributed system.

It is intended for intermediate to advanced developers who are already familiar with the fundamentals of backend development and want to explore concepts beyond the basics.

The project demonstrates a production-inspired architecture using Docker for containerization, Maven multi-module project structure for code organization, and service-to-service communication between independent services. It is designed to provide hands-on experience with building, running, and managing multiple services in a scalable backend environment.

Project Anatomy :

Project Folder Structure

What are covered :

⚡ Event-Driven Architecture
📡 Server-Sent Events (SSE)
📨 Kafka
🐳 Docker
🧩 Multi-Module Architecture
🌐 API Gateway

yes this episode will give you very brief about each one, but trust me going this is approach like ladder principle instead of learning each topic in depth just learn that much that just let you move in and later we make it more robust and secure.

Before we start building, let's first understand what an event-driven system is and why it has become a fundamental architectural pattern for modern distributed applications. We'll explore how services communicate through events, why this approach improves scalability and flexibility, and compare the different technologies available for implementing event-driven architectures.

What Is an Event-Driven System?

Before we start building, let's first understand what an event-driven system is and why it's widely used in modern distributed applications.

In an event-driven architecture, services communicate by producing and consuming events instead of calling each other directly. When something important happens—such as a user placing an order or a payment being completed—an event is published. Any interested service can then react to that event independently.

This approach makes systems more loosely coupled, scalable, resilient, and easier to extend as new services are added.

What Are Our Options?

There are several ways to implement event-driven communication, each suited to different use cases:

Apache Kafka – A distributed event streaming platform designed for high throughput, durability, and scalability.
RabbitMQ – A message broker that excels at reliable message delivery and task distribution.
Redis Pub/Sub – A lightweight publish-subscribe mechanism for simple, low-latency messaging.
Cloud Messaging Services – Managed solutions such as AWS SNS/SQS, Google Pub/Sub, and Azure Service Bus.

Throughout this series, we'll use Apache Kafka to build our event-driven system and understand why it's one of the most popular choices for scalable microservices.

OK OK.. Since now we have to deal with Kafka, let's understand about it from the surface level to gets started and later we will use it at it's full potential..maybe😅.
I have provide some diagrams and notes for each sub-topics you can use that as well in the diagrams and notes folder in Github Repo.
GitHub Link
Let's understand some terminologies associated with Kafka to get into it. You can also refer this diagram to get some high-level understanding first.

Terms associated with Kafka

Producer : A Producer publishes events to Kafka.A producer never knows who will consume the event.
Consumer : A Consumer reads events from Kafka and reacts to them.One event can trigger multiple independent services.
Event : An event represents something that has happened in your system.examples : UserRegistered,OrderPlaced etc.
Topic : A Topic is a logical category where related events are stored.Examples we place all order related events in order-service topic, shipment related events in the shipment-service topic and so on.
Message (Record) : A Message (also called a Record) is one event stored inside a topic.Each message contains key, value, timestamp and offset.
Partition : A topic is divided into partitions. The reason kafka does to achieve parallel processing, higher throughput and scalability.
Ordering is guaranteed only within a partition.
Offset : Every message inside a partition has an Offset.Offsets uniquely identify messages within a partition.Consumers use offsets to know where to resume reading.
Broker : A Broker is a Kafka server.A Kafka cluster consists of one or more brokers.
Cluster : A Cluster is a collection of brokers working together.Adding brokers improves scalability and fault tolerance.Every broker in a Kafka cluster knows about all the other brokers in that same cluster, they do this by maintaining up-to-date cluster metadata, which contains the unique IDs, hostnames, and ports of every single broker, as well as which broker is currently the leader for any given topic partition.
Consumer Group : Multiple consumers can belong to the same Consumer Group.Kafka distributes partitions among consumers so each message is processed only once per group.

Replication : Kafka copies partitions across brokers.
Key : The Key determines which partition stores a message.All events with the same key go to the same partition, preserving order.
Value : The Value is the actual event payload.
Serialization : Kafka transmits bytes.Objects must be converted into bytes before sending.
*Event Streaming *: Kafka continuously streams events.Unlike traditional messaging, Kafka retains events, allowing consumers to replay or process them later.

I think🤔 this must be good enough to sail our ship properly.Let's talk about the framework we are going to use and that willllllll.... beeee... our enterprise king Spring Boot👑.

Let's see how we are going to design since we are planned to create multiple modular maven project.So for this we have parent pom file that contains all the things and create docker docker compose file for running all the services. you can again check my project link and look for the docker compose services Docker file for reference. Link

Let's create events for that will be sending to Kafka topics since you Kafka accepts bytes so we have to serialize and consumer have to deseialize it.

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ShipmentStartedEvent {
    private String orderId;
    private String shipmentId;
    private String customerId;
    private String status;
    private BigDecimal totalAmount;
    private String shippingStreet;
    private String shippingCity;
    private String shippingState;
    private String shippingZip;
    private long timestamp;
}

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class DeliveryStartedEvent {
    private String deliveryId;
    private String orderId;
    private String shipmentId;
    private String customerId;
    private String status;
    private BigDecimal totalAmount;
    private String shippingStreet;
    private String shippingCity;
    private String shippingState;
    private String shippingZip;
    private long timestamp;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class OrderCreatedEvent {
    private String orderId;
    private String customerId;
    private BigDecimal totalAmount;
    private String shippingStreet;
    private String shippingCity;
    private String shippingState;
    private String shippingZip;
    private long timestamp;
}

We setup topics and other config related stuff in the application.properties or application.yaml.

For sending events to Kafka topics with data we can create some Beans in Spring to create KafkaTemplate which we use to send events first or act as event producer.

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, DeliveryStartedEvent> deliveryStartedEventProducerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, DeliveryStartedEvent> deliveryKafkaTemplate() {
        return new KafkaTemplate<>(deliveryStartedEventProducerFactory());
    }
}

Now we can rely on the Spring framework that it will add this as dependancy so that we can use in our services.

private final KafkaTemplate<String, DeliveryStartedEvent> kafkaTemplate;
public DeliveryService(DeliveryRepository deliveryRepository,
                           KafkaTemplate<String, 

DeliveryStartedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

kafkaTemplate.send(deliveryEventsTopic, deliveryEvent.getDeliveryId(), deliveryEvent);

Now similarly we can create Beans for the Kafka Consumers to listen whenever Kafka topic receives event.

@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id:delivery-service-group}")
    private String groupId;

    @Bean
    public ConsumerFactory<String, Object> shipmentStartedEventConsumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.tracker.common_events");
        config.put(JsonDeserializer.TYPE_MAPPINGS,
                "com.tracker.common_events.ShipmentStartedEvent:com.tracker.common_events.ShipmentStartedEvent");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> shipmentStartedEventListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(shipmentStartedEventConsumerFactory());
        factory.setConcurrency(2);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
        backOff.setMaxElapsedTime(30000L);
        factory.setCommonErrorHandler(new DefaultErrorHandler(backOff));

        return factory;
    }
}

factory.setConsumerFactory(orderCreatedEventConsumerFactory());

Connecting to the Broker -> It injects a configuration blueprint (the ConsumerFactory) into your listener container factory.

factory.setConcurrency(3);

It tells Spring to spin up exactly 3 background consumer threads inside this single application instance.

factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

It completely disables Kafka's default "auto-commit" behavior, forcing the application to take manual control over when a message is marked as "read."
Why this is critical for Production: By default (enable.auto.commit = true), Kafka automatically commits messages periodically in the background, even if your Java code throws an error or crashes halfway through processing. This can result in permanently lost events.

With AckMode.MANUAL, Spring Boot stops auto-committing. It waits until your listener method finishes executing without throwing any exceptions. Once the method completes successfully, the framework automatically signals Kafka to commit that specific offset. If your code fails, no commit happens, protecting your data.

ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
backOff.setMaxElapsedTime(30000L);

This instantiates an algorithmic delay calculator used when handling unexpected application failures.

How the math works: * 1000L: The initial interval. If your code fails, it will wait 1 second (1000ms) before trying again.

2.0: The multiplier. If the first retry fails, it multiplies the previous wait time by 2.0.
setMaxElapsedTime(30000L): The global time limit. It states that the loop can continue backing off and retrying, but the cumulative time spent retrying this single failing message cannot exceed 30 seconds.

factory.setCommonErrorHandler(new DefaultErrorHandler(backOff));

What it does: It registers a global error handler that wraps around your listener methods, utilizing the exponential back-off rules defined above.

Under the Hood: If your consumer throws an exception (like a database timeout or a network failure), the DefaultErrorHandler catches it before it can crash the background thread.

It pauses the consumer loop, rolls back the uncommitted manual offset, waits for the calculated back-off duration, and then feeds the exact same message back into your listener method. If the message still fails after the 30-second maximum elapsed time limit is reached, the DefaultErrorHandler logs a critical error and moves on to the next message, preventing a single "poison pill" from stopping your entire pipeline forever.

Now In order to listen we can make use of spring-boot-starter-kafka the KafkaListener annotation to listen to topics.

@KafkaListener(
            topics = "${app.kafka.topics.order-events:order-events}",
            containerFactory = "orderCreatedEventListenerFactory",
            groupId = "${spring.kafka.consumer.group-id:shipment-service-group}"
    )
    public void onOrderCreated(
            @Payload OrderCreatedEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.info("Received OrderCreatedEvent: key={}, partition={}, offset={}, orderId={}, customerId={}",
                key, partition, offset, event.getOrderId(), event.getCustomerId());

        try {
            shipmentService.createShipment(event);
            ack.acknowledge();
            log.debug("Acknowledged offset={} for orderId={}", offset, event.getOrderId());
        } catch (Exception e) {
            log.error("Failed to process OrderCreatedEvent for orderId={}: {}", event.getOrderId(), e.getMessage(), e);
            throw e;
        }
    }

You must have have Headers and Acknowledge in the function parameters
Acknowledge is used for Prevent Data Loss : You ensure a message is only committed after it is safely saved to your database.Error Handling: If your system crashes mid-process, the unacknowledged message will be re-delivered upon restart.

Header values are generated automatically by the Apache Kafka broker and the Spring Kafka framework during the message lifecycle.

Now at this moment we know how to produce events to Kafka topics and how Consumer can get those.

Right now, each of our microservices is exposed through its own URL. As the number of services grows, do you think it's a good idea for our frontend or clients to know the address of every single service?

So far, each microservice in our system is exposed through its own URL. While this is fine for development, it doesn't scale well. Clients need to know the address of every service, making the system tightly coupled and harder to maintain.

To solve this, we'll introduce an API Gateway. Instead of communicating with individual services, clients will send all requests to a single entry point. The gateway will handle request routing while also centralizing concerns like authentication, rate limiting, logging, and load balancing.

Let's integrate an API Gateway and simplify our architecture.
we are going to use another Spring related dependancy in order to accomplish this :

 <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-gateway-server-webflux</artifactId>
 </dependency>

Now we can use the RouteLocator Bean to resolve all microservices and now we access all of them using single address.

@Bean
    public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
        return builder.routes()
                .route("events-stream-java", r -> r
                        .path("/api/v1/events/stream", "/api/v1/events/stream/**")
                        .filters(f -> f.setResponseHeader("Cache-Control", "no-cache"))
                        .uri("http://notification-service:8080"))
                .route("customer-service", r -> r
                        .path("/api/v1/users/**", "/api/v1/orders/**")
                        .uri("http://customer-service:8080"))
                .route("shipment-service", r -> r
                        .path("/api/v1/shipments/**")
                        .uri("http://shipment-service:8080"))
                .route("delivery-service", r -> r
                        .path("/api/v1/deliveries/**")
                        .uri("http://delivery-service:8080"))
                .build();
    }

This diagram can be the overall overview of what we have done so far with the help of the docker compose we wrap everything.

You can find the docker-compose file in the project folder, here

Now, I think we have covered everything to get started and yeah this not perfect but its good enough to think or it gave lot of space to make us think how we can improve this further.

A Note on Using AI

Throughout this series, I encourage you to use AI as a learning and productivity tool—but not as a replacement for engineering judgment.

AI can help you write code faster, explain concepts, and even suggest implementations. However, designing scalable systems still requires critical thinking. Decisions around architecture, scalability, security, reliability, maintainability, and trade-offs cannot simply be delegated to an AI.

It's also important not to accept every piece of generated code without understanding it. As engineers, we're responsible for keeping our codebase clean, organized, and maintainable. That means following SOLID principles, applying appropriate design patterns, avoiding code duplication, reducing tight coupling, and writing code that's easy to understand and evolve.

Think of AI as a co-worker, not the architect. It performs best when given clear context and well-defined problems, but the responsibility for making sound engineering decisions always remains with us.

So, as we build this project together, let's use AI to accelerate our workflow while continuing to think critically and make deliberate design decisions.

Now that we've put together our event-driven architecture, our services can automatically react to events and publish them to any interested consumers.

To see this in action, let's build a simple Server-Sent Events (SSE) endpoint. This will allow clients to receive real-time updates whenever new events are published.

If you're new to Server-Sent Events (SSE), here's a quick overview before we dive into the implementation.

Server-Sent Events (SSE) is a web technology that allows a server to push continuous, real-time updates to a client over a single, long-lived HTTP connection.

Since our current architecture (which isn't perfect yet 😄—but we'll improve it step by step) already includes a Notification Service, it's the ideal place to expose an SSE endpoint. Any client that wants to receive real-time notifications can simply subscribe to this endpoint, and the server will push updates whenever new events occur.

@Service
public class SseEmitterService {

    private static final Logger log = LoggerFactory.getLogger(SseEmitterService.class);
    private final CopyOnWriteArrayList<SseEmitter> emitters = new CopyOnWriteArrayList<>();

    public SseEmitter createEmitter() {
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        emitters.add(emitter);
        emitter.onCompletion(() -> emitters.remove(emitter));
        emitter.onTimeout(() -> emitters.remove(emitter));
        emitter.onError(e -> emitters.remove(emitter));
        log.debug("SSE client connected, total clients={}", emitters.size());
        return emitter;
    }

    public void broadcast(String eventType, Object data) {
        for (SseEmitter emitter : emitters) {
            try {
                emitter.send(SseEmitter.event()
                        .name(eventType)
                        .data(data));
            } catch (IOException e) {
                emitters.remove(emitter);
                log.debug("SSE client disconnected, total clients={}", emitters.size());
            }
        }
    }
}

We can now call this broadcast method on the every KafkaListener and can send events like this :

sseEmitterService.broadcast("DeliveryStatusChanged", event);
sseEmitterService.broadcast("OrderCreated", event);
sseEmitterService.broadcast("ShipmentStatusChanged", event);

As always, you can explore the implementation in the project repository to see how everything fits together in a real-world example.link

With this, we've made a lot of progress—even if it looks a little unstructured right now. 😄 But that's how every great system begins: a collection of ideas that gradually evolve into something well-designed through consistency, iteration, and continuous improvement.

In the next episode, we'll refine our architecture, improve the design, and take another step toward building a production-ready system.

See you in the next episode! 🚀