DEV Community: Farhan Khan

Testcontainers

Farhan Khan — Mon, 26 Jan 2026 22:11:14 +0000

1. Introduction

Testing code that depends on databases and external services is difficult. Many teams rely on shared test environments or in-memory databases like H2, but both approaches are error-prone. Shared environments break due to data conflicts or configuration changes, while in-memory databases often behave differently from production systems.

Testcontainers solves this by running real dependencies inside Docker containers during tests. Each test run gets its own isolated, disposable environment that closely matches production. This is especially valuable for database integration testing, where using the actual database engine leads to more reliable tests and fewer production surprises.

2. Docker

Let’s start with a quick refresher on Docker containers. This article assumes you already have some familiarity with Docker and its core components. That said, we’ll briefly revisit the high-level concepts to align terminology and build a shared mental model before moving forward.

Docker file

A Dockerfile defines the steps required to build a Docker image. For programmers, a useful way to think about a Dockerfile is as a class definition. It acts as a template from which Docker Engine creates a Docker image, and that image can then be used to create multiple container instances, similar to creating objects from a class. Docker Engine processes the Dockerfile and builds a Docker image, which can be executed anywhere as long as the host system uses a compatible operating system kernel.

A Dockerfile allows you to:

Choose a base image
Install system packages and dependencies
Copy application files
Set environment variables and configuration
Expose ports
Define entry points

Docker Image

A Docker image is a packaged snapshot of an application and everything it needs to run, including required libraries, dependencies, and a minimal base filesystem.

Docker Engine creates the image by processing a Dockerfile. Once built, the image is immutable, meaning it never changes, and all containers created from it behave consistently.

Docker images are not heavyweight because they do not include a full operating system or kernel. Instead, they use a minimal filesystem and a layered design, where layers are shared and reused across images to keep them small and efficient.

Docker Container

A Docker container is a running instance of a Docker image, created and started by Docker Engine.

Containers provide an isolated environment for applications. Each container has its own filesystem, process space, network interfaces, and environment variables. Containers are isolated from one another, so processes in one container cannot see or interfere with processes in another.

At the same time, containers share the host operating system kernel. This shared kernel is what makes containers lightweight compared to virtual machines.

Containers are mutable at runtime, but any changes made inside a container are discarded when the container is removed unless explicitly persisted using volumes.

3. Testcontainers

How Testcontainers Works

At a high level, Testcontainers starts Docker containers as part of your test execution.

When a test begins, Testcontainers pulls the required Docker image (if it’s not already available), starts a container, and waits until the service inside the container is ready to accept connections. Your test code then connects to this container using dynamically provided hostnames, ports, and credentials.

Once the tests finish, Testcontainers automatically stops and removes the containers. This ensures each test run starts from a clean, isolated state with no leftover data or configuration.
All of this happens programmatically inside your test framework, so developers do not need to manually start databases or manage shared environments. The same tests can be run locally and in CI with consistent behavior, as long as Docker is available.

Here are some helpful Docker commands when working with Testcontainers:

# Verify Docker is running
docker info

# Pull an image manually (example: MySQL)
docker pull mysql:8.0

# List downloaded MySQL images
docker images | grep mysql

MySQL Integration Test

Gradle dependencies:

dependencies {
    testImplementation "org.junit.jupiter:junit-jupiter:5.10.1"

    // Testcontainers core + JUnit 5 support
    testImplementation "org.testcontainers:testcontainers:1.19.7"
    testImplementation "org.testcontainers:junit-jupiter:1.19.7"

    // MySQL Testcontainer
    testImplementation "org.testcontainers:mysql:1.19.7"

    // MySQL JDBC driver
    testImplementation "mysql:mysql-connector-j:8.3.0"
}

test {
    useJUnitPlatform()
}

Integration test

import org.junit.jupiter.api.Test;
import org.testcontainers.containers.MySQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

import static org.junit.jupiter.api.Assertions.assertEquals;

@Testcontainers
class MySqlIT {

  @Container
  static final MySQLContainer<?> mysql =
      new MySQLContainer<>("mysql:8.0")
          .withDatabaseName("test_db")
          .withUsername("test")
          .withPassword("test");

  @Test
  void canInsertAndQuery() throws Exception {
    try (Connection conn =
             DriverManager.getConnection(mysql.getJdbcUrl(), mysql.getUsername(), mysql.getPassword());
         Statement stmt = conn.createStatement()) {

      stmt.execute("CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(100))");
      stmt.execute("INSERT INTO users (id, name) VALUES (1, 'FK')");

      try (ResultSet rs = stmt.executeQuery("SELECT name FROM users WHERE id = 1")) {
        rs.next();
        assertEquals("FK", rs.getString(1));
      }
    }
  }
}

Kafka Integration Test

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

import static org.junit.jupiter.api.Assertions.assertTrue;

@Testcontainers
class KafkaIT {

  @Container
  static final KafkaContainer kafka =
      new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.3"));

  @Test
  void canProduceAndConsume() {
    String topic = "events";

    // Producer
    Properties producerProps = new Properties();
    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
      producer.send(new ProducerRecord<>(topic, "k1", "hello-from-testcontainers"));
      producer.flush();
    }

    // Consumer
    Properties consumerProps = new Properties();
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "it-group");
    consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
      consumer.subscribe(Collections.singletonList(topic));

      boolean found = false;
      long deadlineMs = System.currentTimeMillis() + 5_000;

      while (!found && System.currentTimeMillis() < deadlineMs) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(250));
        found = records.records(topic).stream()
            .anyMatch(r -> "hello-from-testcontainers".equals(r.value()));
      }

      assertTrue(found, "Expected message was not consumed within timeout");
    }
  }
}

Core Concepts of Kafka

Farhan Khan — Thu, 02 Oct 2025 01:17:10 +0000

1. Introduction

Apache Kafka is a distributed event streaming platform. It’s designed to handle large volumes of real-time data, enabling applications to publish, subscribe to, store, and process streams of events (messages). Think of it as a central nervous system for data in modern systems.

2. Core Concepts

Producer

A Producer is a client application that writes data into Kafka. It creates messages, prepares them for sending, and delivers them to Kafka brokers. Most of its responsibilities can be configured based on reliability, latency, and throughput requirements.

Send messages to Kafka brokers.
Create messages as ProducerRecord objects (with topic, key, value, etc.).
Serialize keys and values into bytes.
Select partition (either user-specified or chosen by the partitioner).
Batch messages together for efficiency.
Handle retries if a send fails.
Receive acknowledgments from the broker (success or error).
Return metadata (topic, partition, offset) when a message is successfully written.

Consumer

A consumer reads messages from Kafka topics. If one consumer alone can’t keep up with the incoming data, Kafka lets you add more consumers in the form of a consumer group.

Within a consumer group, the topic’s partitions are split among the consumers:

If the group has one consumer, it reads from all partitions.
If there are as many consumers as partitions, each consumer gets one partition.
If there are more consumers than partitions, the extras stay idle.

This is how Kafka scales consumption — by letting multiple consumers share the load across partitions. It’s also why topics are often created with multiple partitions, so more consumers can be added later if traffic grows.

When multiple applications need the same data, each application should have its own consumer group. That way, each group gets a full copy of all the messages in the topic, independent of other groups.

Partition

A partition in Kafka is a way of splitting a topic into smaller parts so messages can be stored and processed in parallel. When producing messages, you create a ProducerRecord with a topic, a value, and optionally a key. The key is important because, by default, Kafka will hash the key so that all messages with the same key always go to the same partition. This ensures ordering — for example, all transactions of a single bank account can be kept in sequence. If you don’t provide a key, Kafka uses a round-robin (or “sticky round-robin” in newer versions) to spread messages evenly across partitions. In some cases, the default isn’t enough, so you can implement custom partitioning — for example, sending all records for a very large customer to a dedicated partition to avoid overloading one partition. The main advantage of partitions is that they let Kafka scale: data is distributed across brokers for higher throughput, while still keeping strict ordering for messages that share the same key.

For example, imagine a topic called CustomerOrders that has 4 partitions. If a producer sends a ProducerRecord("CustomerOrders", "Alice", "Order123"), Kafka will hash the key "Alice" and always send her orders to the same partition. Another customer, "Bob", might be mapped to a different partition, and if a record is sent without a key, Kafka will assign it round-robin across the available partitions. This way, the topic can handle thousands of customers at once by spreading the load across partitions, while still keeping each customer’s orders in order.

3. Configuration

Configuring Producer

Following are some important properties to configure Producer

bootstrap.servers
This is a list of broker addresses (host:port) that the producer uses to connect to Kafka for the first time. You don’t need to list all brokers — the producer will discover the rest. It’s best to list at least two, so if one broker is down, the producer can still connect.

key.serializer
This setting tells the producer how to turn the record key into bytes, because Kafka only understands byte arrays. For example, StringSerializer converts a String into bytes, and IntegerSerializer converts an Integer into bytes. Kafka already includes serializers for many common types, so in most cases you don’t need to write your own. The setting is required even if you don’t plan to use keys, but in that case you can use the VoidSerializer.

value.serializer
The same way you set key.serializer to a name of a class that will serialize the message key object to a byte array, you set value.serializer to a class that will serialize the message value object.

acks
The acks setting decides how many brokers must confirm a message before the producer considers it successfully sent. This affects both reliability and speed.

acks=0
The producer doesn’t wait for any reply. Messages can be lost, but sending is very fast.

acks=1
The producer gets confirmation once the leader broker has the message. Safer than 0, but messages can still be lost if the leader crashes before replication.

acks=all
The producer waits until all in-sync replicas (ISR) have the message. Safest option, since the data survives broker crashes, but it’s also the slowest. But how many replicas must be “in-sync” is controlled by min.insync.replicas (a broker or topic-level setting). If min.insync.replicas=2, then at least 2 replicas (leader + 1 follower) must acknowledge the message before it is considered successful.

retries
The retries setting controls how many times the producer will try again if sending a message fails due to a temporary error (like no leader for a partition). The wait time between retries is set by retry.backoff.ms (default 100 ms).
Retries can cause out-of-order messages, because later messages may succeed while earlier ones are still retrying. To avoid this, set max.in.flight.requests.per.connection=1, but this lowers throughput.
With idempotence enabled (enable.idempotence=true) makes sure retries don’t create duplicate messages. Without it, a lost acknowledgment can cause the producer to resend the same record, and the broker will store it twice. With idempotence on, Kafka detects duplicates and only keeps one copy.

max.in.flight.requests.per.connection
max.in.flight.requests.per.connection controls how many batches a producer can send before getting replies from the broker. A higher value improves throughput but uses more memory.
By default it’s 5. But if retries are enabled and this is more than 1, messages can get delivered out of order — because a later batch might succeed while an earlier one is still retrying.
To keep both performance and ordering, it’s best to use enable.idempotence=true, which prevents duplicates and preserves order even with multiple in-flight requests.

enable.idempotence
When retries happen, the same message can sometimes be written more than once, creating duplicates. Setting enable.idempotence=true prevents this by adding a sequence number to each message so the broker can ignore duplicates. It also ensures that messages stay in the correct order, even when retries occur.
It requires acks=all, retries > 0, and max.in.flight.requests.per.connection ≤ 5.

Configuring Consumer

max.poll.records
This setting controls how many records a consumer can get in a single call to poll(). It lets you limit the number of messages your application processes at once, but it doesn’t control their total size.

max.poll.interval.ms
This sets the maximum time a consumer can go without calling poll(). If this time is exceeded, Kafka assumes the consumer is stuck and removes it from the group, triggering a rebalance. The default is 5 minutes. It acts as a safeguard to detect consumers that are not actually processing messages, even if they are still sending heartbeats in the background.

enable.auto.commit
This setting decides if Kafka should automatically save the consumer’s progress (offsets). By default, it’s true, so offsets are committed for you at regular intervals (controlled by auto.commit.interval.ms). The risk is that offsets might be committed before processing finishes, which can cause data loss if the app crashes.

If you set it to false, you commit offsets manually, usually after processing is complete. This avoids missing data, though it may cause duplicates if a message is reprocessed after a crash. To handle duplicates safely, many applications use idempotent writes (e.g., database operations that ignore or update existing records instead of inserting them twice).

partition.assignment.strategy
This setting decides how Kafka will split topic partitions among consumers in the same group. Kafka uses a PartitionAssignor class to do this, and you can choose between several built-in strategies (or write your own).

Range – gives each consumer a block of consecutive partitions per topic. It’s simple, but if partitions don’t divide evenly, some consumers may get more partitions than others.

RoundRobin – spreads partitions one by one across consumers. This usually results in a more balanced distribution.

Sticky – aims to balance partitions like RoundRobin, but also tries to keep existing assignments during rebalances so consumers don’t constantly get shuffled around.

Cooperative Sticky – similar to Sticky, but supports cooperative rebalancing, where consumers can continue working on partitions that aren’t reassigned, making rebalances smoother with less interruption.

By default, Kafka uses the Range strategy. You can switch to RoundRobin, Sticky, or CooperativeSticky, or even plug in your own custom assignor class if you need a special partitioning rule.

3. Exactly Once Semantics

Idempotent Producer

The first step toward achieving exactly-once semantics in Kafka is enabling the idempotent producer. Normally, retries can create duplicates if a broker crashes or acknowledges late, causing the producer to resend the same record. By setting enable.idempotence=true, the producer avoids this problem using three pieces of metadata: a producer ID (PID), a sequence number, and an epoch.

The producer ID uniquely identifies the producer instance.
Each record sent to a partition gets a sequence number that increases in order.
The epoch increments whenever the producer restarts, ensuring that old (zombie) producers can’t write again.

Brokers track the latest sequence numbers per producer and partition. If they see a duplicate (same PID + sequence number), they reject it; if they see an old epoch, they fence it off. This guarantees that messages are delivered once and in order per partition, even during retries or failures.

Transactions

While the idempotent producer removes duplicates within a single partition, many real-world applications process an input message and then produce results to multiple partitions or topics. To guarantee correctness in this consume–process–produce workflow, Kafka provides transactions. Transactions allow producers to group multiple writes and offset commits into a single atomic unit: either all succeed together or none are applied.

A transactional producer is enabled by setting a transactional.id, which persists across restarts and lets Kafka track producer identity over time. Under the hood, Kafka uses a transaction coordinator that records transaction state in an internal topic (transaction_state). When committing, the coordinator writes commit markers to all involved partitions, ensuring consistency across them. If the producer crashes, a new coordinator can recover the intent from the log and finish the commit or abort, preventing partial results.

From the consumer’s perspective, Kafka provides two isolation levels:

read_uncommitted (default): consumers see all records, including those from aborted or still-open transactions.

read_committed: consumers only see records that are part of committed transactions, ensuring that partial or aborted writes are never exposed. This mode provides the exactly-once guarantee, but it may introduce slight lag since consumers must wait until a transaction is finalized.

Together, transactional producers and read_committed consumers provide true exactly-once semantics across partitions and topics.

Consensus in Distributed Databases

Farhan Khan — Tue, 23 Sep 2025 03:46:13 +0000

1. Introduction

Consensus is the backbone of fault-tolerant distributed systems, ensuring that nodes agree on a single outcome even in the face of crashes, delays, or network partitions. Without consensus, systems risk split-brain scenarios, inconsistent reads, and conflicting writes. Algorithms like Paxos, Raft, and Zab power consensus in real-world infrastructure such as ZooKeeper, etcd, Consul, Spanner, and CockroachDB. By doing so, they provide the foundation not only for strong consistency models like linearizability but also for the coordination primitives that keep large clusters safe and reliable.

Properties of Consensus Algorithm

From a theoretical perspective, consensus algorithms have three properties:

Agreement – The decision value is the same for all correct processes.
Validity – The decided value was proposed by one of the processes.
Termination – All correct processes eventually reach the decision.

Applications of Consensus

Some of the most important applications of consensus include:

Leader election – ensures exactly one primary node at a time, preventing conflicts and enabling strong consistency models like linearizability.
Replicated logs / state machine replication – guarantees a single, agreed-upon order of operations across replicas, forming the backbone of reliable database replication.
Locks and leases – allows only one process to safely acquire a critical resource, even under failures or network partitions.
Uniqueness constraints – prevents multiple clients from successfully creating conflicting records (e.g., the same username or primary key).
Atomic transaction commit – ensures that all participants in a distributed transaction either commit or abort together, avoiding partial results.
Shard and workload assignment – coordinates which nodes own which shards or tasks, and automatically reassigns them when nodes join, leave, or fail.
Globally ordered ID generation – provides unique, monotonic IDs that can serve as sequence numbers, event ordering markers, or tickets.

2. Shared Logs: Consensus in Practice

A shared log is an abstraction where many nodes or clients can propose entries, and the system ensures every replica sees the same append-only sequence in the same order. In practice, a single leader (at a time) appends entries; consensus decides the order and guarantees all replicas deliver the same history.

A shared log is also formally known as a total order broadcast, atomic broadcast, or total order multicast protocol. These terms describe the same idea in different words: requesting a value to be added to the log is called broadcasting, and reading a log entry is called delivering.

Properties of a Shared Log

Eventual append – if a node proposes an entry and does not crash, it will eventually see that entry appear in the log.
Reliable delivery – once any correct node sees an entry, every correct node will eventually see it.
Append-only – entries are immutable; new entries can only be added after existing ones.
Agreement (total order) – any two nodes that have read entry must have read exactly the same prefix before, in the same order.
Validity – every log entry came from a valid proposal (no fabricated entries).

Most practical consensus protocols (Raft, Multi-Paxos, Zab) expose exactly this log interface.

3. Consensus Algorithms (Implementations)

While consensus can be described abstractly, real systems rely on specific algorithms to make it work. The three most influential and widely used are Paxos, Raft, and Zab. Each solves the consensus problem, but with different trade-offs in understandability, performance, and adoption.

Paxos

Developed by Leslie Lamport; the first practical consensus algorithm.
Works through two phases of voting: first, nodes agree on a proposal number, and second, they agree on the actual value.
Guarantees safety even in the face of failures, but the logic is subtle and the algorithm is notoriously hard to implement correctly.
Multi-Paxos extends the protocol to agree on a sequence of values, effectively forming a shared log.

Raft

Designed with understandability in mind, making it simpler than Paxos.
Elects a leader through a voting process with term numbers.
The leader appends client requests to its log, replicates them to followers, and considers entries committed once a majority acknowledge them.
Ensures that a newly elected leader has an up-to-date log before continuing, simplifying failover.
Widely used in systems like etcd and Consul.

Zab

Stands for Zookeeper Atomic Broadcast.
Specifically designed to support ZooKeeper, a coordination service.
Uses a leader-based approach where the leader proposes updates, and followers acknowledge before the update is committed.
Emphasizes total order broadcast, ensuring every update is applied in the same sequence across all replicas.
Provides strong consistency guarantees for coordination tasks like locks and configuration management.

We’ll only touch on Paxos, Raft, and Zab here. Each will be covered in more detail in separate articles; this section is just a high-level overview of how consensus works in practice.

Core Concepts of MCP

Farhan Khan — Mon, 15 Sep 2025 03:54:12 +0000

1. Introduction

The Model Context Protocol (MCP) is an open standard that lets AI systems connect to external tools and data in a consistent way. An MCP server exposes capabilities such as resources (data), tools (actions), and prompts (templates), while clients like IDEs or AI agents can automatically discover and use them.

Built on JSON-RPC 2.0, MCP is transport-agnostic and modular, so a server written once can work with any compatible client. This standardization makes AI integrations more scalable, reusable, and secure compared to custom one-off APIs.

2. Protocol and Transport

MCP is built on JSON-RPC 2.0, a lightweight standard for structured request–response communication. JSON-RPC provides a predictable format for requests, responses, and errors, making MCP simple to implement across programming languages.

A major design choice is that MCP is transport-agnostic. The protocol defines what the messages look like, but not how they are delivered. This means the same MCP server can communicate over:

stdin/stdout (common for local developer tools)
sockets (for long-lived processes)
web transports such as WebSockets (for cloud-hosted or browser-based integrations)

This separation of protocol and transport ensures MCP integrations are portable and future-proof.

Example: JSON-RPC Request in MCP

Request

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "callTool",
  "params": {
    "name": "validateConfig",
    "arguments": {
      "filePath": "configs/sample.pkl"
    }
  }
}

Response

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "status": "ok",
    "issues": []
  }
}

3. Resources

Resources are a core building block of an MCP server. They represent structured data that the server makes available for clients to read. A resource could be a configuration file, a schema, a database table, or any other external dataset that the server wants to expose.

Each resource comes with metadata (such as its name, description, and MIME type) and a schema that defines the shape of the data. This ensures that clients know exactly what structure to expect when they request the resource.
Importantly, the MCP protocol itself defines standard methods for working with resources:

resources/list – lets clients discover what resources exist (metadata only).
resources/read – lets clients fetch the actual contents of a specific resource.

This standardization is what makes MCP interoperable: clients don’t need custom code for each server. They just call the same protocol methods, and the server fills in the details about its resources.
By enforcing this contract, MCP allows clients to interact with external data in a predictable and reusable way — without ad hoc APIs.

Discovery

Agents typically follow a Retrieval-Augmented Generation (RAG) pattern when working with resources:

Retrieve – Call resources/list (or a search tool) to identify candidate resources using metadata like title, description, mimeType, or schema.
Read – Fetch selected resources with resources/read to get their actual content.
Augment – Insert the retrieved content into the model’s prompt/context to provide grounding knowledge.
Generate – Produce the final answer, explanation, or action using both the user’s query and the augmented data.
This makes resources the knowledge base layer of MCP, ensuring that agents can dynamically pull in just the data they need to reason effectively.

Scalability

When a server exposes a large number of resources (thousands or even millions), MCP provides mechanisms to keep discovery and retrieval efficient:

Pagination – resources/list supports cursors, so clients can fetch resources in manageable chunks.
Resource templates – Servers can advertise URI templates (e.g., product://{category}/{id}) so clients can generate URIs without enumerating every possible item.
Server-side search tools – Instead of scrolling huge lists, clients can call a search tool (like searchCatalog) to fetch a small set of relevant URIs.
Subscriptions – Clients can subscribe to changes instead of constantly re-listing, making updates lightweight.

By combining these strategies, MCP ensures that agents scale gracefully: they don’t need to load everything, just the right resources at the right time.

Example: Resource Interaction

Step 1 — Client lists resources


{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "resources/list",
  "params": {}
}

Response

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resources": [
      {
        "uri": "product://starship/ftl-2000",
        "name": "FTL-2000",
        "title": "FTL-2000 Starship",
        "description": "Faster-than-light exploration vessel with warp drive",
        "mimeType": "application/json",
        "schema": {
          "type": "object",
          "properties": {
            "id": { "type": "string" },
            "price": { "type": "number" },
            "currency": { "type": "string" },
            "specs": {
              "type": "object",
              "properties": {
                "engine": { "type": "string" },
                "crewCapacity": { "type": "integer" },
                "range": { "type": "string" }
              },
              "required": ["engine", "crewCapacity"]
            }
          },
          "required": ["id", "price", "currency", "specs"]
        }
      }
    ]
  }
}

Step 2 – Client requests a product resource


{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/read",
  "params": {
    "uri": "product://starship/ftl-2000"
  }
}

Response

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "contents": [
      {
        "uri": "product://starship/ftl-2000",
        "name": "FTL-2000",
        "title": "FTL-2000 Starship",
        "mimeType": "application/json",
        "text": "{ \"id\": \"ftl-2000\", \"price\": 1200000000, \"currency\": \"Galactic Credits\", \"specs\": { \"engine\": \"Warp Core MkIV\", \"crewCapacity\": 12, \"range\": \"Andromeda Sector\" } }"
      }
    ]
  }
}

3. Tools

Tools are a core building block of an MCP server. If Resources are the data layer, Tools are the action layer: operations the server can perform on behalf of the client/agent (for example: compareProducts, checkInventory, manageShoppingCart).
MCP standardizes how clients discover and run tools:

tools/list — discover available tools, their parameters, and result schemas (metadata only).
callTool — invoke a specific tool with structured arguments; receive a structured result. Tools always define: name (unique identifier) description (what it does) input schema (arguments in JSON Schema) output schema (result in JSON Schema)

Tools always define:

name – unique identifier)
description – what it does
input schema – arguments in JSON Schema
output schema – result in JSON Schema

This makes tool use predictable, typed, and interoperable across different clients.

Discovery

Before a tool can be invoked, it must first be discovered. The client calls tools/list, and the server responds with metadata describing each available tool: its name, description, input schema, and output schema.
This step is crucial because it makes MCP dynamic:
The client doesn’t need to know in advance what tools a server supports.
The server can add or remove tools without requiring client changes.
The schemas tell the client exactly what arguments to pass and what result format to expect.
Once discovered, tools can be invoked immediately using callTool.

Example: Compare Products Tool

Step 1 – Discover tool

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

Response

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "compareProducts",
        "description": "Compare two product JSON specs and return key differences.",
        "inputSchema": {
          "type": "object",
          "properties": {
            "left":  { "type": "object" },
            "right": { "type": "object" }
          },
          "required": ["left", "right"]
        },
        "outputSchema": {
          "type": "object",
          "properties": {
            "differences": {
              "type": "array",
              "items": {
                "type": "object",
                "properties": {
                  "path": { "type": "string" },
                  "left": {},
                  "right": {}
                },
                "required": ["path", "left", "right"]
              }
            }
          },
          "required": ["differences"]
        }
      }
    ]
  }
}

Step 2 – Call the tool

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "callTool",
  "params": {
    "name": "compareProducts",
    "arguments": {
      "left":  { "id": "ftl-2000", "specs": { "range": "Andromeda", "crewCapacity": 12 } },
      "right": { "id": "quantum-comm", "specs": { "range": "Galaxy-wide", "crewCapacity": 0 } }
    }
  }
}

Response

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "differences": [
      { "path": "specs.range",        "left": "Andromeda",   "right": "Galaxy-wide" },
      { "path": "specs.crewCapacity", "left": 12,            "right": 0 }
    ]
  }
}

5. Prompts

Prompts are a core building block of an MCP server. If Resources are the data layer and Tools are the action layer, Prompts are the guidance layer. They are reusable prompt templates that servers expose to clients, helping agents interact with models in a consistent and structured way.
Prompts can be used to:

Standardize how models summarize or explain data.
Provide reusable templates for common tasks (e.g., summarizeTechSpecs, trainingIntro, generateAdCopy).
Reduce prompt-engineering overhead by centralizing phrasing in the server.

Discovery

Prompts are discovered dynamically. The client calls prompts/list, and the server responds with metadata describing each available prompt:

name: unique identifier of the prompt
description: what the prompt is for
arguments: placeholders that can be filled in by the client

Because prompts live on the server, they can be added, updated, or removed without changing client logic. This ensures agents always use the most up-to-date phrasing patterns.
Servers may also expose prompts in different personas or styles, allowing the same data to be presented in very different ways. For example:

A strictInstructor prompt could return concise, no-nonsense training commands for starship crews.
An enthusiasticSalesRep prompt could turn the very same specs into energetic sales copy designed to excite customers.

This flexibility means the agent can pick the right “voice” depending on whether the user wants discipline or persuasion.

Example: Prompt Discovery

Imagine the user says to the agent:

User> I need both a technical spec summary and a sales pitch for the FTL-2000 starship.

Step 1: Client lists prompts

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "prompts/list",
  "params": {}
}

Response

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "prompts": [
      {
        "name": "strictInstructor",
        "description": "Provide concise, no-nonsense training instructions.",
        "arguments": [{ "name": "topic", "required": true }]
      },
      {
        "name": "enthusiasticSalesRep",
        "description": "Pitch products with energy and persuasive flair.",
        "arguments": [{ "name": "productSpec", "required": true }]
      }
    ]
  }
}

Step 2: Client fetches and uses enthusiasticSalesRep

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "prompts/get",
  "params": {
    "name": "enthusiasticSalesRep",
    "arguments": {
      "productSpec": "{ \"id\": \"ftl-2000\", \"specs\": { \"engine\": \"Warp Core MkIV\", \"crewCapacity\": 12, \"range\": \"Andromeda Sector\" } }"
    }
  }
}

Response

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "messages": [
      {
        "role": "user",
        "content": "Write an energetic, persuasive pitch for this product:\n\n{ \"id\": \"ftl-2000\", \"specs\": { \"engine\": \"Warp Core MkIV\", \"crewCapacity\": 12, \"range\": \"Andromeda Sector\" } }"
      }
    ]
  }
}

Step 3: Model Output

User> I need both a technical spec summary and a sales pitch for the FTL-2000 starship.

LLM> Step aboard the FTL-2000 Starship — the ultimate vessel for bold explorers! With a Warp Core MkIV engine, room for 12 adventurers, and the range to reach Andromeda, this ship isn’t just transportation, it’s freedom across the stars.

6. MCP Agent Workflow

We’re showing how an agent uses MCP to answer a user request about the FTL-2000 starship, producing both a tech spec summary and a sales pitch.

Step 1 — User Request
The user says:

User> Give me a technical spec summary and a sales pitch for the FTL-2000 starship.

Step 2 — Discovery

The agent is discovering available prompts, resources, and tools it can use to answer the user’s request.

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "prompts/list",
  "params": {}
}

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/list",
  "params": { "filters": { "product": "FTL-2000" } }
}

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/list",
  "params": {}
}

Step 3 — Acquisition

The agent reads the relevant resources (e.g., technical spec file and marketing notes for the FTL-2000) to gather the factual content it will use.

{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "resources/read",
  "params": { "uri": "res://catalog/starships/FTL-2000/spec.json" }
}

{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "resources/read",
  "params": { "uri": "res://marketing/starships/FTL-2000/benefits.md" }
}

Step 4 — Prompt Composition
The agent fills in two prompt templates: one for a technical summary and one for a sales pitch, injecting the resource data it just gathered.

{
  "jsonrpc": "2.0",
  "id": 6,
  "method": "prompts/get",
  "params": {
    "name": "techSpecSummarizer",
    "arguments": {
      "subject": "FTL-2000",
      "context": {
        "spec": {
          "drive": "Quantum-Entangled Fold Drive v3",
          "range_ly": 4200,
          "crew": 8,
          "cargo_tons": 120,
          "power_core": "Tri-helix fusion",
          "safety_rating": "A+"
        }
      }
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 7,
  "method": "prompts/get",
  "params": {
    "name": "salesPitchPersona",
    "arguments": {
      "product": "FTL-2000",
      "audience": "Fleet procurement",
      "facts": "Half the fuel, double the range; A+ safety; 15-minute turnaround"
    }
  }
}

Step 5 — Generation
The agent uses the resolved prompts with its model and produces two outputs:

Technical Spec Summary (FTL-2000): Range 4,200 ly; crew 8; cargo 120 tons; drive: Quantum-Entangled Fold v3; power: Tri-helix fusion; safety: A+.

Sales Pitch (Fleet Procurement): Cut fuel in half, double the range. A+ safety. 15-minute turnaround keeps ships earning.

Vector Database: Core Concepts

Farhan Khan — Thu, 04 Sep 2025 06:35:31 +0000

1️⃣ Introduction

🔹 Vector

A vector is simply an ordered list (array) of numbers.
Can represent data points in 2D, 3D, or higher dimensions.
Example:
- 2D → [3.5, 7.2]
- 3D → [1.2, -4.5, 6.0]
In machine learning, vectors are used to describe positions in a multi-dimensional space.

🔹 Embedding Vector

An embedding vector (or just embedding) is a special kind of vector generated by an embedding model.
Purpose: represent complex data (text, images, audio, etc.) in a way that captures meaning and similarity.
All embeddings are vectors, but not all vectors are embeddings.
Dimensions (e.g., 384, 768, 1536) are fixed by the model design.
- Lightweight (384–512 dimensions) → e.g., all-MiniLM-L6-v2 (384d), Universal Sentence Encoder (512d)
- Standard (768–1,536 dimensions) → e.g., all-mpnet-base-v2 (768d), OpenAI text-embedding-ada-002 (1,536d)
- High capacity (2,000–3,000+ dimensions) → e.g., OpenAI text-embedding-3-large (3,072d)
Higher dimensions = richer detail, but more costly to store and search.
Represents the semantic meaning of data, so similar concepts are placed close together in vector space.
- “dog” and “puppy” → embedding vectors close together.
- “dog” and “car” → embedding vectors far apart.

🔹 Positional Encoding

Positional encoding is a technique used in Transformer models to provide word order information to embeddings.
Purpose: ensure that the sequence of tokens matters, so sentences with the same words in different orders have different meanings.
Implemented by adding a positional vector to each token embedding before feeding it into the model.
- Can be sinusoidal (fixed mathematical functions).
- Or learned (trainable position embeddings).
Formula (sinusoidal encoding):
- PE(pos, 2i) = sin(pos / 10000^(2i/d_model))
- PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model))
Sentence-level embeddings indirectly include positional information, since it’s baked into the Transformer during encoding.
Preserves contextual meaning by distinguishing different word orders.
- “dog bites man” → one meaning.
- “man bites dog” → very different meaning.

🔹 Vector Databases

A vector database is a system built to store and retrieve vectors, especially embeddings.
Core features:
- Persistence: store large volumes of embeddings.
- Similarity search: find nearest neighbors using cosine similarity, dot product, or Euclidean distance.
- Indexing: HNSW, IVF, Annoy, PQ for fast retrieval.
- Database functions: CRUD operations, filtering, replication, and scaling.
Purpose: enable semantic search → finding results based on meaning instead of exact keyword matches.

2️⃣ Persistence

Vectors often number in the millions or billions, so keeping them only in memory is not practical.
Persistence ensures embeddings are stored long-term and survive restarts or failures.
CRUD: create, read, update, and delete vectors.
Durability: vectors are written to disk or distributed storage.
Index persistence: not just raw vectors, but also the indexing structures (like HNSW graphs).

🔹 Example: creating an embedding (OpenAI)

from openai import OpenAI
client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="The cat sat on the mat"
)

vector = response.data[0].embedding
print(len(vector))  # 1536 dimensions

🔹 Example: insertion (ElasticSearch)

PUT /documents/_doc/1
{
  "content": "The cat sat on the mat",
  "embedding": [0.12, -0.87, 0.33]
}

PUT /documents/_doc/2
{
  "content": "The dog chased the ball",
  "embedding": [0.91, 0.04, -0.22]
}

3️⃣ Similarity Search

Core function of a vector database → find vectors closest in meaning to a query vector.
Powers semantic search, recommendations, fraud detection, and Retrieval-Augmented Generation (RAG).

🔹 Common distance metrics

Cosine similarity: measures angle between vectors.
Euclidean distance: straight-line distance in vector space.
Dot product: measures alignment, often used when vectors are normalized.

🔹 Example: semantic search (Elasticsearch)

k-Nearest Neighbors (kNN) is the task of finding the k most similar vectors to a query vector. In vector databases, it’s the core operation behind semantic search, recommendations, and RAG.

POST /documents/_search
{
  "size": 2,
  "knn": {
    "field": "embedding",
    "query_vector": [0.15, -0.52, 0.48, ...],
    "k": 2,
    "num_candidates": 10
  },
  "_source": ["content"]
}

4️⃣ Indexing

Searching vectors directly without an index is too slow for large datasets.
Indexing structures organize vectors so that nearest-neighbor queries can be answered efficiently.
These methods implement Approximate Nearest Neighbor (ANN) search → trading a bit of accuracy for major speed gains.

🔹 Popular ANN-based algorithms

Approximate Nearest Neighbor (ANN) search is a family of algorithms that find the closest vectors quickly by trading a small amount of accuracy for massive gains in speed and scalability — the following algorithms are all ANN-based.

HNSW (Hierarchical Navigable Small World Graph)

Builds a multi-layer graph where each vector connects to its neighbors.
✅ Fast queries, high recall, supports dynamic updates.
❌ High memory consumption, complex to tune.

IVF (Inverted File Index)

Clusters vectors into groups (using k-means or similar).
At query time, only the most relevant clusters are searched.
✅ Efficient for large datasets, reduces search scope.
❌ Accuracy depends on clustering quality.

PQ (Product Quantization)

Compresses vectors into compact codes to save memory.
✅ Great for billion-scale datasets, reduces storage dramatically.
❌ Some loss in accuracy due to compression.

Annoy (Approximate Nearest Neighbors)

Builds multiple random projection trees for searching.
✅ Lightweight, simple to use, good for read-heavy static datasets.
❌ Slower than HNSW at very large scale, poor for frequent updates.

ScaNN (Scalable Nearest Neighbors, Google)

Optimized for high-dimensional, large-scale data.
✅ Very fast, optimized for Google-scale workloads, low memory footprint.
❌ Less community support, limited flexibility outside Google’s ecosystem.

🔹 Trade-offs

Speed vs. Memory: Some indexes (like HNSW) are fast but memory-hungry.
Accuracy vs. Compression: PQ saves space but reduces precision.
Dynamic vs. Static Data: HNSW handles updates well, while Annoy is better for static data.

5️⃣ Filtering & Metadata

Pure similarity search often returns results that are semantically close but not contextually relevant.
Real-world apps combine semantic similarity with structured filters (e.g., category, date, user ID).
Example: “find similar support tickets from the past month” or “recommend products in the Shoes category.”

🔹 How it works

Each vector is stored with metadata → key-value pairs like:
- category: "electronics"
- created_at: "2025-09-01"
- user_id: 12345
At query time, the DB runs a hybrid search: 1) Apply metadata filters. 2) Run similarity search only on the filtered subset.

🔹 Getting the query vector (two common options)

from openai import OpenAI
client = OpenAI()

q = "Wireless noise-cancelling headphones"
emb = client.embeddings.create(
    model="text-embedding-3-small",
    input=q
).data[0].embedding  # length = 1536

🔹 Example with metadata filter (Elasticsearch)

POST /documents/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "knn": {
            "embedding": {
              "vector": [/* paste emb here, e.g., 0.12, -0.87, 0.33, ... */],
              "k": 3
            }
          }
        },
        { "term": { "category": "electronics" } }
      ]
    }
  }
}

A2A (Agent to Agent): Core Concepts

Farhan Khan — Fri, 29 Aug 2025 04:02:21 +0000

1️⃣ Introduction

Agent-to-Agent (A2A) Protocol is an open standard that enables independent AI agents to discover each other, share capabilities, and collaborate on tasks in a secure and interoperable way. Instead of building one monolithic agent, A2A allows specialized agents—such as a planner, researcher, or executor—to communicate seamlessly across different frameworks and organizations. By defining a common structure for discovery (Agent Cards), task management (Task lifecycle), and communication (messages & parts), A2A provides the foundation for scalable, multi-agent ecosystems where agents can work together much like microservices do in modern software architectures.

2️⃣ Agent Cards

A small, self-describing JSON document that advertises an agent’s identity, capabilities, and how to interact with it.
It typically lives at a well-known URL like https://yourdomain.com/.well-known/agent.json (or agent-card.json).
It enables automated discovery and interoperability so other agents can safely submit tasks and receive updates without custom integrations.
A consuming agent fetches and parses the card → learns endpoints, formats, and auth → submits tasks accordingly.

🔹 Agent Card Keys

id

A globally unique identifier (URN/UUID) for the agent (e.g., urn:agent:example:researcher-1).

name

Human-readable name for UIs and logs.

description

Short summary of the agent’s purpose and domain.

version

Semantic version of the agent service; helps with change management and client compatibility.

endpoints/task_submit

HTTP endpoint where other agents POST tasks for execution.
Accepts a task object (JSON). Returns task acceptance/ID and may start processing asynchronously.

endpoints/events

Endpoint to stream progress updates (SSE/webhooks) or poll for status/results.
Clients subscribe or query using the previously returned task ID.

capabilities

A list describing actions/skills (e.g., web.search, summarize).
Lets peers quickly assess whether this agent can fulfill a requested task.

modalities

Supported input/output formats (e.g., text/plain, application/json).
Ensures payload compatibility between agents.

auth

Required authentication scheme(s) (e.g., OAuth2 scopes, mTLS, API keys).
Enforces secure, auditable access aligned with enterprise policies.

skills (optional)

Richer, structured descriptions of capabilities with tags and examples.
Improves automated routing/selection among multiple candidate agents.

protocolVersion (optional)

The A2A protocol version implemented.
Helps clients negotiate behavior and avoid drift.

🔹 Example Agent Card

{
  "a2a_version": "1.0",
  "id": "urn:agent:example:researcher-1",
  "name": "Researcher Agent",
  "description": "Finds sources on the web and summarizes them.",
  "version": "1.0.0",
  "endpoints": {
    "task_submit": "https://researcher.example.com/a2a/tasks",
    "events": "https://researcher.example.com/a2a/events"
  },
  "capabilities": ["web.search", "summarize"],
  "modalities": ["text/plain", "application/json"],
  "auth": {
    "type": "oauth2",
    "scopes": ["a2a.tasks.write", "a2a.events.read"]
  }
}

3️⃣ Discovery Directory

A shared registry service where agents publish their Agent Cards so others can find them dynamically.
Acts as the DNS + service registry of the A2A ecosystem — but instead of just resolving network addresses, it resolves capabilities.
Allows agents to query by capability, version, or metadata to locate the right peer for a task.
Enables dynamic, loosely coupled ecosystems where agents can join/leave without manual wiring.
ADK provides a Discovery Directory out of the box, so developers can immediately register and look up agents. For advanced use cases, teams can swap in custom registries with richer policies or scaling behavior.

🔹 Discovery Directory Keys

url

Base URL of the directory service (e.g., https://directory.example.com/a2a).
Agents register their cards here and consumers query it for matches.

query

API to request available agents by criteria (capabilities, region, trust level, version).
Returns one or more Agent Cards that match.

registration

Endpoint for an agent to register or update its Agent Card.
Typically authenticated to prevent rogue entries.

metadata

Optional performance/health data stored alongside each entry.
Examples: uptime, SLA, latency profiles, trust scores.
Helps consumers select among multiple candidates.

🔹 Example Directory Query

GET https://directory.example.com/a2a/agents?capability=web.search
Accept: application/json

🔹 Example Directory Response

[
  {
    "id": "urn:agent:example:researcher-1",
    "name": "Researcher Agent",
    "capabilities": ["web.search", "summarize"],
    "endpoints": {
      "task_submit": "https://researcher.example.com/a2a/tasks",
      "events": "https://researcher.example.com/a2a/events"
    },
    "metadata": {
      "region": "us-east1",
      "sla": "99.9%",
      "trustLevel": "verified"
    }
  }
]

4️⃣ Task Submission (A2A)

🔹 Key Points

Submit work to another agent via its task_submit endpoint (from the Agent Card).
Async-first: server returns immediately with { id, status: "submitted", links: { self, events } }.
Track progress using events (SSE/webhook) for real-time updates until terminal status: completed or failed.
Use a client-generated task ID for idempotency; retries with the same ID should not duplicate execution.
Final output is received via the events stream.

🔹 Example Flow (High-Level)

Discover receiver’s task_submit from its Agent Card.
Build task payload: id, requester, and payload (type + args).
POST it to task_submit → receive async ack with links.events.
Subscribe to the events endpoint to get status updates and results.
Print the final status and artifacts when the task is completed.

🔹 Example Task Submit in Python (Async with SSE)

# Brief, happy-path example: async task submission + SSE event subscription
# Focused on clarity, with inline comments explaining each step.

import asyncio, json, uuid, httpx

# Endpoint for submitting tasks — discovered from the Agent Card
TASK_SUBMIT_URL = "https://receiver.example.com/a2a/tasks"

# Access token (OAuth2 bearer, API key, or other scheme from Agent Card auth section)
ACCESS_TOKEN = "<TOKEN>"

async def main():
    # Step 1: Generate a unique task ID for idempotency
    # Using UUID ensures retries with the same ID won't duplicate execution if the server deduplicates
    task_id = f"task-{uuid.uuid4()}"

    # Step 2: Build the payload
    # Includes:
    # - id: chosen by the client
    # - requester: who is submitting (URN or URL)
    # - payload: actual work request (type + arguments)
    payload = {
        "id": task_id,
        "requester": "urn:agent:example:planner-1",
        "payload": {
            "type": "web.search",
            "query": "Latest AI conferences in 2025"
        }
    }

    async with httpx.AsyncClient() as client:
        # Step 3: Submit the task
        # POST returns quickly with async acknowledgment (id, status=submitted, links)
        ack = (await client.post(
            TASK_SUBMIT_URL,
            headers={
                "Authorization": f"Bearer {ACCESS_TOKEN}",  # auth as required by Agent Card
                "Accept": "application/json"                # we want JSON back
            },
            json=payload,
        )).json()

        print("Submitted", ack["id"], "status=", ack["status"])
        # ack contains "links": { "self": ..., "events": ... }
        # - links.self: poll for updates
        # - links.events: subscribe for streaming updates (preferred)

        # Step 4: Subscribe to events stream (SSE)
        # Server sends progress + final results as "data: {...}" lines
        events_url = ack["links"]["events"]
        async with client.stream(
            "GET",
            events_url,
            headers={
                "Authorization": f"Bearer {ACCESS_TOKEN}",
                "Accept": "text/event-stream"
            }
        ) as resp:
            # Loop over incoming SSE lines until we see a terminal status
            async for line in resp.aiter_lines():
                if not line or not line.startswith("data:"):
                    continue  # ignore comments/keepalives
                evt = json.loads(line[5:].strip())  # strip "data:" prefix, parse JSON
                if "status" in evt:
                    print("[event] status=", evt["status"])
                    # Break once we hit completed/failed — final output is available
                    if evt["status"] in {"completed", "failed"}:
                        final = evt
                        break
                else:
                    # Could be partial result or log message
                    print("[event]", evt)

    # Step 5: Print final output
    print("Final status:", final["status"])
    for a in final.get("artifacts", []):
        # Artifacts may have inline content (text) or links (href)
        print("-", a.get("type"), a.get("content", a.get("href", "<no content>")))

if __name__ == "__main__":
    asyncio.run(main())

🔹 Example Output

Submitted: {"id": "task-123", "status": "submitted", "links": {"events": "..."}}
[event] {"status": "working"}
[event] {"status": "completed", "artifacts": [{"type": "text/plain", "content": "Upcoming AI conferences: NeurIPS 2025, ICML 2025..."}]}
Final output: {"status": "completed", "artifacts": [{"type": "text/plain", "content": "Upcoming AI conferences: NeurIPS 2025, ICML 2025..."}]}

5️⃣ Messages & Parts

🔹 Key Points

Messages are a general-purpose communication channel between agents, outside of task submission.
Each message has metadata (id, sender, recipient, timestamp) and a list of parts.
Typed payloads such as text, JSON, image, or file references. Each part has a type and either content or href.
Share context, logs, prompts, human-in-the-loop requests, or supporting artifacts.
Messages are POSTed to the messages endpoint advertised in the Agent Card.
Typed parts let heterogeneous agents parse what they understand and ignore the rest.

🔹 Example Flow

Agent A POSTs a message to Agent B’s messages endpoint.
The message contains metadata plus one or more typed parts.
Agent B processes each part as appropriate (display, parse JSON, fetch URI, etc.).

🔹 Example Message (JSON)

{
  "id": "msg-42",
  "sender": "urn:agent:example:planner-1",
  "recipient": "urn:agent:example:researcher-1",
  "timestamp": "2025-08-28T12:00:00Z",
  "parts": [
    {
      "type": "text/plain",
      "content": "Please summarize this document."
    },
    {
      "type": "application/json",
      "content": { "docId": "abc123", "priority": "high" }
    },
    {
      "type": "text/uri-list",
      "href": "https://example.com/docs/abc123.pdf"
    }
  ]
}

🔹 Example Server Response

{
  "id": "msg-3f1e3f7b-2c58-42ab-94cd-fc22e3b1687a",
  "status": "ok",
  "received_at": "2025-08-28T12:34:56Z",
  "echo": {
    "parts": [
      { "type": "text/plain", "content": "Please summarize this document." },
      { "type": "text/uri-list", "href": "https://example.com/docs/abc123.pdf" }
    ]
  }
}

LangGraph: Core Concepts

Farhan Khan — Thu, 28 Aug 2025 20:55:59 +0000

LangGraph is a framework for building stateful, reliable agent workflows. Instead of executing a single prompt-response cycle, LangGraph enables agents to operate as directed graphs, where each node represents a step, state is shared across the workflow, and execution can branch, pause, or persist.

This article summarizes the core concepts, prebuilt utilities, and typical applications of LangGraph.

1️⃣ State

The memory object that flows through the graph.
Nodes read from and write updates to state.
Managed through channels that define merge strategies:
- Replace → overwrite existing values.
- Append → add items (e.g., chat history).
- Merge → combine dicts/lists.
Acts as the single source of truth for the agent’s execution.

🔹 Example: How State Evolves

Consider a simple graph with three nodes: plan → search → answer.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict, total=False):
    question: str
    plan: str
    results: list[str]
    final_answer: str

def plan_node(state: State) -> State:
    return {"plan": f"Search for: {state['question']}"}

def search_node(state: State) -> State:
    return {"results": [f"Result about {state['plan']}"]}

def answer_node(state: State) -> State:
    return {"final_answer": f"Based on {state['results'][0]}, here is the answer."}

Initial Input

{"question": "What is LangGraph?"}

After plan node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?"}

After search node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?", "results": ["Result about Search for: What is LangGraph?"]}

After answer node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?", "results": ["Result about Search for: What is LangGraph?"], "final_answer": "Based on Result about Search for: What is LangGraph?, here is the answer."}

2️⃣ Nodes

Nodes are the functional units of a LangGraph.
Each node is a Python function that:
- Takes the current state (a dictionary-like object).
- Returns updates to that state (usually as another dictionary).
A node can perform many different actions depending on the workflow:
- Invoke an LLM to generate text, plans, or summaries.
- Call external tools or APIs (e.g., search engine, database, calculator).
- Execute deterministic logic (e.g., scoring, validation, formatting).
Nodes don’t overwrite the whole state by default — instead, they return partial updates that LangGraph merges into the global state using channels.
A node is not an “agent” by itself. The entire graph of nodes forms the agent.

🔹 Example: Simple Node

from typing import TypedDict

class State(TypedDict, total=False):
    question: str
    plan: str

def plan_node(state: State) -> State:
    q = state["question"]
    return {"plan": f"Search online for: {q}"}

Input State

{"question": "What is LangGraph?"}

Output Update

{"plan": "Search online for: What is LangGraph?"}

After merging

{"question": "What is LangGraph?", "plan": "Search online for: What is LangGraph?"}

🔹 Example: Node Invoking an LLM

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def answer_node(state: State) -> State:
    response = llm.invoke(state["question"])
    return {"final_answer": response.content}

👉 Nodes are modular steps. They can be simple (string formatting) or advanced (LLM + tools). Together, they form the workflow.

3️⃣ Edges

Edges define the flow of execution between nodes.
Normal edges → fixed transitions.
Conditional edges → branching logic, using a router function.
Special markers:
- START → entry point.
- END → exit point.

🔹 Example: Normal Edges

from langgraph.graph import StateGraph, START, END

graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "search")
graph.add_edge("search", "answer")
graph.add_edge("answer", END)

🔹 Example: Conditional Edges with Router

def router(state: State) -> str:
    q = state["question"]
    if "latest" in q.lower():
        return "search"
    else:
        return "answer"

graph.add_conditional_edges(
    "plan", router, {"search": "search", "answer": "answer"}
)

Input: "What is the capital of France?" → routed to answer.
Input: "What are the latest news on AAPL?" → routed to search.

4️⃣ Streaming

Streaming provides live feedback during execution.

🔹 Update Stream (node-level)

for event in app.stream(inputs, config=config, stream_mode="updates"):
    print(event)

Output

{'plan': {'plan': 'Search for: What is LangGraph?'}}
{'search': {'results': ['Result about What is LangGraph?']}}
{'answer': {'final_answer': '...final text...'}}

🔹 Token Stream (LLM output)

for chunk in app.stream(inputs, config=config, stream_mode="messages"):
    print(chunk, end="", flush=True)

stream_mode="updates" → node updates.
stream_mode="messages" → token stream (if LLM supports it).

5️⃣ Memory

Memory in LangGraph = state + checkpointers.

🔹 Short-Term Memory (Within a Thread)

from typing import TypedDict, List

class State(TypedDict, total=False):
    question: str
    chat_history: List[str]
    answer: str

def add_to_history(state: State) -> State:
    history = state.get("chat_history", [])
    history.append(state["question"])
    return {"chat_history": history}

Example after two turns:

{"chat_history": ["What is LangGraph?", "Explain checkpointers"], "answer": "..."}

🔹 Long-Term Memory (Across Runs)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

app.invoke({"question": "What is LangGraph?"}, config={"configurable": {"thread_id": "t1"}})
app.invoke({"question": "And what are edges?"}, config={"configurable": {"thread_id": "t1"}})

Both runs share the same thread_id → context is preserved.

🔹 Combined

Short-term memory = within a run.
Long-term memory = across runs.
Together → continuity + reliability.

👉 Unlike LangChain, LangGraph doesn’t treat memory as a separate object — it’s baked into state + checkpointers.