DEV Community: Alain Airom (Ayrom)

Protecting the Crown Jewels: Securing Source Code in Containerized Applications

Alain Airom (Ayrom) — Tue, 14 Apr 2026 09:32:11 +0000

Protect your source code if it’s not open-sourced!

Introduction and some background

In a recent discussion with a technical partner, a challenging question arose: “How can we distribute containerized applications while preventing the source code from being easily viewed or reverse-engineered?” While I had encountered articles on this topic before, I wanted to provide something more tangible than just a list of links. Instead of a simple search, I teamed up with Bob, our expert digital architect, to build a comprehensive demonstration.

Bob provided functional code samples in four different programming languages — Go, C++, Python, and Node.js — to showcase how varying levels of protection can be achieved within a container. The result is a deep dive into the practical strategies you can use to protect your intellectual property when it leaves your internal environment.

What are the Technics that Bob Provided?

The Multi-Stage Advantage: How to ensure your source code never even enters the final production image.
Language-Specific Hardening: From static binary compilation in Go to PyArmor obfuscation in Python.
Minimal Footprints: Using “Distroless” and minimal base images to reduce the attack surface.
Verification: Techniques to prove your source code and build tools have been successfully removed.

Why This Matters?

Shipping a container is often mistaken for shipping a “black box,” but without the right precautions, a simple docker save can reveal your proprietary algorithms. By the end of this guide, you’ll have a roadmap for securing your applications, regardless of whether you are working with compiled or interpreted languages.

Let’s dive into the implementations.

Choosing Your Shield: Language-Specific Protection Strategies

While Bob has provided functional samples in Go, C++, Python, and Node.js, it is important to understand that not all “shields” are created equal. The applications themselves are designed to be simple demonstrations of a complex problem: how to prevent your logic from being reverse-engineered once it’s in the wild.

The Spectrum of Security

Security is rarely bulletproof; it is a series of hurdles designed to make unauthorized access as difficult and expensive as possible. As the comparison below shows, the “right path” depends heavily on the language you choose:

Compiled Languages (Go, C++): These offer the highest level of security. By compiling code into a static binary and stripping debug symbols, the original source code is never present in the container.
Interpreted Languages (Python, Node.js): These are inherently more challenging to protect. Because the runtime requires the source (or bytecode) to function, we must rely on obfuscation — scrambling the code to make it unreadable to humans while remaining executable by the machine.

Key Takeaways for Implementation

Architecture Matters: Protection begins at the conception phase. If an algorithm is truly sensitive, you might choose to implement that specific component in a compiled language like C++ or Go, even if the rest of your stack is in Node.js.
The Effort vs. Security Trade-off: There is often a direct correlation between the effort required to write a language and the security of its output. Low-level languages may require more development time, but they reward you with a significantly more robust “black box” for distribution.
A Solid Guideline: Use these samples not as a final destination, but as a framework. By combining multi-stage builds with minimal images and code-hardening, you create a layered defense that protects your intellectual property from the majority of common threats.

Go Sample Application (one of the good choices)

The Go app is provided below as one of the examples, alongside with the related Dockerfile.

package main

import (
 "crypto/sha256"
 "encoding/hex"
 "fmt"
 "log"
 "net/http"
 "os"
 "time"
)

// Secret business logic - this will be compiled and not visible in the container
const (
 secretKey       = "my-super-secret-algorithm-key-2024"
 licenseCheckURL = "internal-license-server"
)

// ProprietaryAlgorithm represents our "secret sauce"
type ProprietaryAlgorithm struct {
 secretSalt string
 iterations int
}

// NewAlgorithm creates a new instance with proprietary settings
func NewAlgorithm() *ProprietaryAlgorithm {
 return &ProprietaryAlgorithm{
  secretSalt: secretKey,
  iterations: 10000,
 }
}

// ProcessData applies our proprietary algorithm
func (pa *ProprietaryAlgorithm) ProcessData(input string) string {
 // This is our "secret" business logic
 result := input + pa.secretSalt
 for i := 0; i < pa.iterations; i++ {
  hash := sha256.Sum256([]byte(result))
  result = hex.EncodeToString(hash[:])
 }
 return result
}

// ValidateLicense checks if the application is properly licensed
func ValidateLicense() bool {
 // Simulated license validation
 licenseKey := os.Getenv("LICENSE_KEY")
 if licenseKey == "" {
  log.Println("Warning: No license key provided")
  return false
 }

 // In production, this would call a license server
 expectedHash := "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 hash := sha256.Sum256([]byte(licenseKey))
 actualHash := hex.EncodeToString(hash[:])

 return actualHash == expectedHash || licenseKey == "DEMO-LICENSE-KEY"
}

func healthHandler(w http.ResponseWriter, r *http.Request) {
 w.WriteHeader(http.StatusOK)
 fmt.Fprintf(w, `{"status":"healthy","timestamp":"%s"}`, time.Now().Format(time.RFC3339))
}

func processHandler(algo *ProprietaryAlgorithm) http.HandlerFunc {
 return func(w http.ResponseWriter, r *http.Request) {
  if r.Method != http.MethodPost {
   http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
   return
  }

  input := r.URL.Query().Get("data")
  if input == "" {
   http.Error(w, "Missing 'data' parameter", http.StatusBadRequest)
   return
  }

  // Apply our proprietary algorithm
  result := algo.ProcessData(input)

  w.Header().Set("Content-Type", "application/json")
  fmt.Fprintf(w, `{"input":"%s","processed":"%s","algorithm":"proprietary-v1"}`, input, result)
 }
}

func main() {
 // Validate license on startup
 if !ValidateLicense() {
  log.Println("Running in DEMO mode - some features may be limited")
 } else {
  log.Println("License validated successfully")
 }

 // Initialize our proprietary algorithm
 algo := NewAlgorithm()

 // Setup HTTP server
 http.HandleFunc("/health", healthHandler)
 http.HandleFunc("/process", processHandler(algo))

 port := os.Getenv("PORT")
 if port == "" {
  port = "8080"
 }

 log.Printf("Starting Go secure application on port %s", port)
 log.Printf("Endpoints: /health (GET), /process (POST)")

 if err := http.ListenAndServe(":"+port, nil); err != nil {
  log.Fatal(err)
 }
}

// Made with Bob

The Dockerfile

# Multi-stage build for Go application
# Stage 1: Build the application
FROM golang:1.21-alpine AS builder

# Set working directory
WORKDIR /build

# Copy go mod files
COPY go.mod ./

# Download dependencies
RUN go mod download

# Copy source code
COPY main.go ./

# Build the application with optimizations
# -ldflags="-s -w" strips debug information and symbol table
# CGO_ENABLED=0 creates a static binary
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build \
    -ldflags="-s -w -X main.buildTime=$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
    -trimpath \
    -o secure-app \
    main.go

# Stage 2: Create minimal runtime image
# Using distroless for maximum security - no shell, no package manager
FROM gcr.io/distroless/static-debian11:nonroot

# Copy only the binary from builder
COPY --from=builder /build/secure-app /app/secure-app

# Use non-root user (already set in distroless/static:nonroot)
# UID 65532 is the nonroot user

# Expose port
EXPOSE 8080

# Set working directory
WORKDIR /app

# Run the application
ENTRYPOINT ["/app/secure-app"]

# Benefits of this approach:
# 1. Source code is NOT in the final image
# 2. No build tools in the final image
# 3. No shell access (distroless)
# 4. Minimal attack surface
# 5. Binary is stripped of debug symbols
# 6. Non-root user for security
# 7. Image size is minimal (~2MB for Go binary + ~2MB for distroless base)

Test phase;

Analysis and verdict

Python Sample Application (moderate choice)

The Python code

"""
Secure Python Application with Proprietary Algorithm
This code will be obfuscated before containerization
"""

import hashlib
import os
import time
from flask import Flask, request, jsonify

# Secret proprietary constants - will be obfuscated
SECRET_KEY = "python-proprietary-algorithm-2024"
ITERATIONS = 8000


class ProprietaryAlgorithm:
    """
    Our secret business logic class
    This will be obfuscated to protect intellectual property
    """

    def __init__(self):
        self.secret_salt = SECRET_KEY
        self.iterations = ITERATIONS
        self._internal_state = self._initialize_state()

    def _initialize_state(self):
        """Internal initialization - will be obfuscated"""
        return hashlib.sha256(self.secret_salt.encode()).hexdigest()

    def _apply_transformation(self, data: str, iteration: int) -> str:
        """
        Proprietary transformation logic
        This is our "secret sauce" that will be protected
        """
        combined = f"{data}{self._internal_state}{iteration}"
        return hashlib.sha256(combined.encode()).hexdigest()

    def process_data(self, input_data: str) -> str:
        """
        Main processing method with proprietary algorithm
        Multiple iterations make reverse engineering harder
        """
        result = input_data + self.secret_salt

        # Apply our secret transformation multiple times
        for i in range(self.iterations):
            result = self._apply_transformation(result, i)

        return result


class LicenseValidator:
    """License validation logic - will be obfuscated"""

    @staticmethod
    def validate():
        """Check if application is properly licensed"""
        license_key = os.getenv('LICENSE_KEY', '')

        if not license_key:
            print("Warning: No license key provided - running in DEMO mode")
            return False

        # Simulated license validation
        expected_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        actual_hash = hashlib.sha256(license_key.encode()).hexdigest()

        is_valid = actual_hash == expected_hash or license_key == "DEMO-LICENSE-KEY"

        if is_valid:
            print("License validated successfully")
        else:
            print("Invalid license - some features may be limited")

        return is_valid


# Initialize Flask app
app = Flask(__name__)

# Initialize our proprietary algorithm
algorithm = ProprietaryAlgorithm()

# Validate license on startup
license_valid = LicenseValidator.validate()


@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({
        'status': 'healthy',
        'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
        'licensed': license_valid
    })


@app.route('/process', methods=['POST'])
def process_data():
    """Process data using our proprietary algorithm"""
    data = request.args.get('data')

    if not data:
        return jsonify({'error': 'Missing "data" parameter'}), 400

    # Apply our proprietary algorithm
    result = algorithm.process_data(data)

    return jsonify({
        'input': data,
        'processed': result,
        'algorithm': 'python-proprietary-v1'
    })


@app.route('/')
def index():
    """Root endpoint"""
    return jsonify({
        'service': 'Secure Python Application',
        'endpoints': {
            '/health': 'GET - Health check',
            '/process': 'POST - Process data (requires ?data=value)'
        }
    })


if __name__ == '__main__':
    port = int(os.getenv('PORT', 8080))
    print(f"Starting Python secure application on port {port}")
    print(f"Endpoints: /health (GET), /process (POST)")

    # Run with production settings
    app.run(host='0.0.0.0', port=port, debug=False)

# Made with Bob

The related Dockfile

FROM python:3.11-slim AS builder

RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /build

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

RUN pyarmor gen -O dist app.py

RUN echo "Flask==3.0.0" > dist/requirements.txt && \
    echo "Werkzeug==3.0.1" >> dist/requirements.txt

FROM python:3.11-slim

WORKDIR /app

COPY --from=builder /build/dist/requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt && \
    rm requirements.txt

COPY --from=builder /build/dist/ .

RUN useradd -r -u 1001 appuser && \
    chown -R appuser:appuser /app

USER appuser

EXPOSE 8080

ENV PYTHONUNBUFFERED=1
ENV PORT=8080

CMD ["python", "app.py"]

Test phase;

The verdict;

Conclusion
The complete source for these applications is available on GitHub alongside detailed technical guides. However, the true centerpiece of this project is the “Security-Analysis” document.

This comprehensive analysis serves as a strategic roadmap, offering a side-by-side comparison of protection levels, reverse-engineering difficulty, and image-size trade-offs. It doesn’t just show you what was done; it explains the why behind every security layer — from symbol stripping in compiled binaries to control-flow flattening in obfuscated JavaScript.

Whether you are a developer looking for a ‘quick win’ or an architect designing a high-security distribution model, this guide provides the clarity needed to choose the right shield for your intellectual property.

>>> Thanks for reading <<<

The “The Architecture Handbook for Milvus Vector Database” Book Review

Alain Airom (Ayrom) — Sun, 12 Apr 2026 12:20:50 +0000

A feedback and my thoughts on the “The Architecture Handbook for Milvus Vector Database” book.

The image from the book cover produced by Packt Publications

Introduction-Mastering the Core of AI Memory

I think I can say that I know Milvus vector DB; I’ve used it many times locally for testing my applications with my Ollama+LLM+Vector Database, and Milvus is also one of the many vector databases proposed in the watsonx Data platform. However, I found this book very useful, and one of the best parts (among others) is Chapter 16: “Implementing Multi-Tenancy in Milvus”. This handbook, written by the core contributors of the Milvus project, systematically unpacks the full technology stack — from basic concepts and deployment to performance tuning and real-world implementation.

> Disclaimer: This is an independent review. I have no professional or financial relationship with the publisher or the authors.

A synthesis of the Book

Setup and Configuration

Chapter 1: Introduction to Milvus — This chapter sets the stage by defining vector databases and explaining how they handle unstructured data through encoding processes like one-hot and TF-IDF to enable similarity search.
Chapter 2: Deploying Milvus in Multiple Ways — It provides a practical guide to various deployment methods, including Docker Compose, Kubernetes operators, and Helm charts, helping users choose the right environment for their needs.

If you’ve already setup Milvus on your end, you can bypass these first chapters in my opinion!

Chapter 3: Interacting with Milvus — It learns the core objects — fields, schemas, collections, and indexes — and how to interact with the database using both the Python SDK and REST APIs for standard data operations.
Chapter 4: Configuring the Milvus System — This chapter focuses on managing system dependencies and service configurations, including setting up monitoring tools like Prometheus, Grafana, and Loki for log analysis.

Data Model — Reading Data, Modification and Compacting Data

Chapter 5: Understanding the Milvus Data Model and Architecture — It dives into the disaggregated four-tier architecture (access, coordination, worker, and storage) and the “lifeline” of the system: the Message Queue (MQ) and Timetick mechanism.

Chapter 6: Data Modification and Maintenance in Milvus — This section details the lifecycle of data “segments” and the internal journey of insert, delete, and upsert requests through the system’s various layers.

Chapter 7: Reading Data in Milvus — It explains the “scatter-gather” query pattern, replica management, and how Milvus ensures multi-level consistency between data freshness and search latency.
Chapter 8: Compaction and Garbage Collection — This chapter explores how background processes like L0 and clustering compaction optimize storage and query performance by managing fragmented segments and obsolete data.

In the context of the Milvus Vector Database, the processes of Compaction and Garbage Collection (GC) are critical because they address the inherent challenges of a high-performance, distributed LSM-tree (Log-Structured Merge-tree) based system.

These mechanisms are vital for maintaining a healthy and fast vector database:

Maintaining Search Speed (Performance Optimization): When we delete or update data in Milvus, the system doesn’t immediately “scrub” the old data from the disk (as this would be computationally expensive). Instead, it marks data as deleted. Over time, these “logical deletes” create fragmentation. So, during a search, the system still has to scan through these fragmented segments and “skipped” records, which slows down query latency. Therefore the ‘compaction’ merges small segments into larger, more optimized ones and physically removes deleted records, ensuring the search engine only processes relevant, “healthy” data.
Storage Efficiency: Without a ‘Garbage Collection’ system, the storage usage would grow indefinitely even if we dele as much data as we add.
Solving the “Small File” Problem (L0 Compaction): Milvus often receives data in small batches. If left unmanaged, this results in thousands of tiny files (Level 0 segments). Searching across 1,000 tiny files is significantly slower than searching across one large file due to the overhead of opening files and managing metadata. So “Compaction” continuously “rolls up” these tiny L0 segments into larger, indexed segments that are optimized for the Knowhere search engine.
Clustering ‘Compaction for Spatial Locality’: standard compaction just merges files, but Clustering Compaction goes a step further by reorganizing data based on the values of a specific field (like a partition key or a scalar field). By Pruning it allows the system to perform “segment pruning.” If our query only needs data from “Tenant A,” the system can skip entire segments that it knows only contain data for “Tenant B,” drastically reducing the I/O required for a search.
System Stability and Reliability: If segments are never compacted, the system’s metadata (stored in ETCD) becomes bloated. This can lead a) slow startup times, b) increased memory consumption by the Query Nodes and c) potential system crashes when the metadata layer becomes a bottleneck.

Chapter 9: Exploring Milvus’ Vector Engine — The focus shifts to the Knowhere engine, the core computational heart that supports various hardware acceleration frameworks and distance metrics like L2 and Cosine.
Chapter 10: Indexing in Detail: Algorithms and Parameters — (Based on the book’s structure) This chapter likely provides a deep dive into the mathematical foundations of indexing algorithms like HNSW, IVF, and Product Quantization (PQ).
Chapter 11: Performance Testing and Benchmarking — It outlines how to use tools like VectorDBBench to establish performance baselines and simulate data growth to test system limits.
Chapter 12: Stability and Reliability Testing — This chapter focuses on ensuring the system can handle failures through chaos engineering, recovery testing, and stress testing.

Scalabilty and Performance

Chapter 13: Scalability and Capacity Planning — It provides guidelines on how to scale the various tiers of Milvus — query nodes, data nodes, and index nodes — based on specific workload requirements.

Chapter 14: Performance Tuning and Best Practices — This part distills years of engineering experience into actionable advice for optimizing indexing and search parameters.

"""
Chapter 14: Configure in-memory replicas for horizontal scaling.

Loads a collection with a specified number of in-memory replicas to distribute
search load across multiple QueryNode instances.

Usage:
    python load_replicas.py [--uri localhost:19530] [--replicas 3]
"""

import argparse

from pymilvus import MilvusClient

COLLECTION_NAME = "test_collection"


def main():
    parser = argparse.ArgumentParser(description="Configure Milvus in-memory replicas")
    parser.add_argument(
        "--uri", default="http://localhost:19530", help="Milvus server URI"
    )
    parser.add_argument("--token", default="", help="Milvus authentication token")
    parser.add_argument("--collection", default=COLLECTION_NAME, help="Collection name")
    parser.add_argument(
        "--replicas", type=int, default=3, help="Number of in-memory replicas"
    )
    args = parser.parse_args()

    client = MilvusClient(uri=args.uri, token=args.token)

    # Release collection first if already loaded
    try:
        client.release_collection(args.collection)
        print(f"Released collection '{args.collection}'")
    except Exception:
        pass

    # Load with specified replica count
    client.load_collection(
        collection_name=args.collection,
        replica_number=args.replicas,
    )
    print(
        f"Collection '{args.collection}' loaded with {args.replicas} in-memory replicas"
    )
    print(
        f"Search requests will be distributed across {args.replicas} QueryNode instances"
    )


if __name__ == "__main__":
    main()

Advanced Topics

Chapter 15: Advanced Search and Querying — It explores techniques like batch search processing and horizontal scaling with multiple replicas to handle thousands of concurrent users.

"""
Chapter 15: Batch Search Optimization

This example demonstrates how batch processing dramatically improves
search throughput by amortizing overhead costs using real Wikipedia embeddings.
"""

import os
import time
from datasets import load_dataset
from pymilvus import MilvusClient, DataType


def load_wikipedia_data(limit=50000, batch_size=1000):
    """
    Load Wikipedia embeddings from Hugging Face dataset.

    Args:
        limit: Number of records to load
        batch_size: Records per batch

    Yields:
        Batch of records formatted for Milvus
    """
    print(f"Loading Wikipedia dataset (total_records={limit})...")
    docs = load_dataset("Cohere/wikipedia-22-12-simple-embeddings", split="train")

    # Use slice to get data efficiently
    for start_idx in range(0, limit, batch_size):
        end_idx = min(start_idx + batch_size, limit)
        batch_docs = docs[start_idx:end_idx]

        # Format batch for Milvus
        batch = []
        for i in range(len(batch_docs["emb"])):
            batch.append(
                {
                    "emb": batch_docs["emb"][i],
                    "title": batch_docs["title"][i][:500],  # Truncate to max length
                    "text": batch_docs["text"][i][:2000],  # Truncate to max length
                    "wiki_id": batch_docs["wiki_id"][i],
                    "views": float(batch_docs["views"][i]),
                }
            )

        yield batch


def get_search_vectors(num_searches=100):
    """Get real search vectors from Wikipedia dataset."""
    print(f"Loading {num_searches} search vectors from dataset...")
    docs = load_dataset("Cohere/wikipedia-22-12-simple-embeddings", split="train")

    search_vectors = []

    for i, doc in enumerate(docs):
        if i >= num_searches:
            break
        search_vectors.append(doc["emb"])

    return search_vectors


def main():
    # Connect to Milvus
    client = MilvusClient(uri=os.getenv("MILVUS_URI", "http://localhost:19530"))

    collection_name = "wiki_articles"
    num_vectors = 50000
    dim = 768

    # ========================================================================
    # Setup: Create Collection and Insert Data
    # ========================================================================
    print("=" * 70)
    print("Setup: Creating collection and inserting data")
    print("=" * 70)

    if client.has_collection(collection_name):
        client.drop_collection(collection_name)

    # Create collection
    schema = client.create_schema(auto_id=True, enable_dynamic_field=False)
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="emb", datatype=DataType.FLOAT_VECTOR, dim=dim)
    schema.add_field(field_name="title", datatype=DataType.VARCHAR, max_length=100)
    schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=5000)
    schema.add_field(field_name="wiki_id", datatype=DataType.INT64)
    schema.add_field(field_name="views", datatype=DataType.FLOAT)

    client.create_collection(collection_name=collection_name, schema=schema)

    # Create HNSW index
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="emb",
        index_type="HNSW",
        metric_type="COSINE",
        params={"M": 16, "efConstruction": 200},
    )
    client.create_index(collection_name=collection_name, index_params=index_params)

    # Insert data in batches
    batch_num = 0
    total_inserted = 0
    for batch in load_wikipedia_data(limit=num_vectors, batch_size=1000):
        client.insert(collection_name=collection_name, data=batch)
        batch_num += 1
        total_inserted += len(batch)
        if batch_num % 10 == 0:
            print(f"Inserted {total_inserted} records...")

    print(f"Total inserted: {total_inserted} records")

    # Load collection
    client.load_collection(collection_name)

    search_params = {"metric_type": "COSINE", "params": {"ef": 128}}

    # ========================================================================
    # Example 1: Single Search
    # ========================================================================
    print("\n" + "=" * 70)
    print("Example 1: Single Search (Sequential)")
    print("=" * 70)

    search_vectors = get_search_vectors(num_searches=100)
    print("Running 100 sequential single searches...")

    start_time = time.time()
    for i, search_vector in enumerate(search_vectors):
        client.search(
            collection_name=collection_name,
            data=[search_vector],
            anns_field="emb",
            search_params=search_params,
            limit=10,
        )
    total_time = time.time() - start_time

    print(f"Total time: {total_time:.2f}s")
    print(f"Average time per search: {total_time / len(search_vectors) * 1000:.2f}ms")
    print(f"Throughput: {len(search_vectors) / total_time:.2f} searches/sec")

    # ========================================================================
    # Example 2: Batch Search
    # ========================================================================
    print("\n" + "=" * 70)
    print("Example 2: Batch Search (All at Once)")
    print("=" * 70)

    print("Running single batch search with 100 searches...")

    start_time = time.time()
    results = client.search(
        collection_name=collection_name,
        data=search_vectors,
        anns_field="emb",
        search_params=search_params,
        limit=10,
    )
    total_time = time.time() - start_time

    print(f"Total time: {total_time:.2f}s")
    print(f"Average time per search: {total_time / len(search_vectors) * 1000:.2f}ms")
    print(f"Throughput: {len(search_vectors) / total_time:.2f} searches/sec")

    print("\nSample results from first search:")
    for i, hit in enumerate(results[0][:5]):
        print(f"  {i + 1}. ID: {hit['id']}, Distance: {hit['distance']:.4f}")

    # ========================================================================
    # Cleanup
    # ========================================================================
    print("\nCleaning up...")
    client.release_collection(collection_name)
    client.drop_collection(collection_name)
    print("Done!")


if __name__ == "__main__":
    main()

Chapter 16: Implementing Multi-Tenancy in Milvus — This standout chapter details different strategies for isolating tenant data, ranging from database-level isolation to collection-level and partition-key-based approaches.

Database-Level Multi-Tenancy in Milvus
This notebook demonstrates how to implement database-level isolation for multiple tenants using Milvus. Each tenant gets their own dedicated database, providing the strongest form of tenant isolation.

Prerequisites
Running Milvus instance (v2.5.x or later)
PyMilvus SDK (v2.5.8 or compatible)
Setup and Configuration
# Install required packages if not already installed
# !pip install pymilvus==2.5.8
from pymilvus import MilvusClient, DataType
import random

# Configuration
MILVUS_URI = "http://localhost:19530"  # Update this to your Milvus instance
client = MilvusClient(uri=MILVUS_URI)

print(f"Connected to Milvus at {MILVUS_URI}")
Define Tenant Configurations
Each tenant will have their own database with specific collections.

# Define tenant configurations
TENANTS = {
    "company_a": {"db_name": "tenant_company_a", "collections": ["products", "users"]},
    "company_b": {
        "db_name": "tenant_company_b",
        "collections": ["inventory", "customers"],
    },
    "company_c": {
        "db_name": "tenant_company_c",
        "collections": ["documents", "analytics"],
    },
}

print(f"Configured {len(TENANTS)} tenants:")
for tenant_id, config in TENANTS.items():
    print(
        f"  - {tenant_id}: {config['db_name']} with collections {config['collections']}"
    )
Tenant Database Setup Functions
def setup_tenant_database(tenant_id, config):
    """Set up a dedicated database for a tenant"""
    db_name = config["db_name"]

    # Create database for tenant
    try:
        client.create_database(db_name)
        print(f"✓ Created database '{db_name}' for {tenant_id}")
    except Exception as e:
        print(f"Database '{db_name}' may already exist: {e}")

    # Switch to tenant database
    client.using_database(db_name)

    # Create collections for tenant
    for coll_name in config["collections"]:
        schema = client.create_schema(auto_id=True, enable_dynamic_fields=True)

        # Add fields (example schema)
        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=768)
        schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1000)
        schema.add_field(
            field_name="category", datatype=DataType.VARCHAR, max_length=100
        )

        # Create collection
        try:
            client.create_collection(collection_name=coll_name, schema=schema)
            print(f"✓ Created collection '{coll_name}' in '{db_name}'")

            # Create index for vector field
            client.create_index(
                collection_name=coll_name,
                field_name="vector",
                index_params={"index_type": "FLAT", "metric_type": "COSINE"},
            )
            print(f"✓ Created index for '{coll_name}'")

        except Exception as e:
            print(f"Collection '{coll_name}' may already exist: {e}")

    # Switch back to default database
    client.using_database("default")
    return True
def insert_sample_data(tenant_id, collection_name, num_records=100):
    """Insert sample data into a tenant's collection"""
    db_name = TENANTS[tenant_id]["db_name"]

    # Switch to tenant database
    client.using_database(db_name)

    # Generate sample data
    data = []
    categories = ["electronics", "books", "clothing", "home", "sports"]

    for i in range(num_records):
        data.append(
            {
                "vector": [random.random() for _ in range(768)],
                "text": f"{tenant_id} item {i + 1}",
                "category": random.choice(categories),
            }
        )

    # Insert data
    client.insert(collection_name=collection_name, data=data)

    # Load collection for searching
    client.load_collection(collection_name)

    print(f"✓ Inserted {num_records} records into {tenant_id}/{collection_name}")

    # Switch back to default database
    client.using_database("default")
    return True
def query_tenant_data(tenant_id, collection_name, query_vector, limit=5):
    """Query data from a specific tenant's database"""
    db_name = TENANTS[tenant_id]["db_name"]

    # Switch to tenant database
    client.using_database(db_name)

    # Perform search - data is completely isolated
    results = client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=limit,
        output_fields=["text", "category"],
    )

    # Switch back to default database
    client.using_database("default")

    return results
Setup All Tenant Databases
Create separate databases for each tenant with their specific collections.

# Setup databases for all tenants
print("Setting up tenant databases...")
for tenant_id, config in TENANTS.items():
    print(f"\nSetting up {tenant_id}:")
    setup_tenant_database(tenant_id, config)
Insert Sample Data for Each Tenant
# Insert sample data for each tenant
print("Inserting sample data...")
for tenant_id, config in TENANTS.items():
    print(f"\nInserting data for {tenant_id}:")
    for collection_name in config["collections"]:
        insert_sample_data(tenant_id, collection_name, num_records=50)
Demonstrate Tenant Isolation
Show that each tenant can only access their own data.

# Generate a query vector
query_vector = [random.random() for _ in range(768)]

print("Demonstrating tenant isolation:")
print("\nQuerying Company A's products:")
results_a = query_tenant_data("company_a", "products", query_vector, limit=3)
for i, result in enumerate(results_a[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )

print("\nQuerying Company B's inventory:")
results_b = query_tenant_data("company_b", "inventory", query_vector, limit=3)
for i, result in enumerate(results_b[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )

print("\nQuerying Company C's documents:")
results_c = query_tenant_data("company_c", "documents", query_vector, limit=3)
for i, result in enumerate(results_c[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )
Verify Database Isolation
# List all databases to verify isolation
databases = client.list_databases()
print("All databases in Milvus:")
for db in databases:
    print(f"  - {db}")

print(
    f"\nTotal tenant databases created: {len([db for db in databases if db.startswith('tenant_')])}"
)
Test Cross-Tenant Access (Should Fail)
Demonstrate that tenants cannot access each other's data.

# Try to access Company B's data from Company A's database context
print("Testing cross-tenant access restriction:")
try:
    # Switch to Company A's database
    client.using_database(TENANTS["company_a"]["db_name"])

    # Try to access Company B's collection (this should fail)
    results = client.search(
        collection_name="inventory",  # Company B's collection
        data=[query_vector],
        limit=5,
    )
    print("❌ Cross-tenant access succeeded (this shouldn't happen!)")
except Exception as e:
    print(f"✓ Cross-tenant access properly blocked: {str(e)[:100]}...")
finally:
    # Switch back to default database
    client.using_database("default")
Performance and Resource Analysis
# Analyze resource usage per tenant
print("Database-Level Multi-Tenancy Analysis:")
print("\nBenefits:")
print("  ✓ Maximum data isolation (100% secure)")
print("  ✓ Simple tenant lifecycle management")
print("  ✓ Granular RBAC control per database")
print("  ✓ Zero risk of data leakage between tenants")

print("\nLimitations:")
print("  ⚠ Limited to ~64 tenants (Milvus database limit)")
print("  ⚠ 40-60% higher memory usage per tenant")
print("  ⚠ 3-5x more operational complexity")
print("  ⚠ Higher infrastructure costs")

total_collections = sum(len(config["collections"]) for config in TENANTS.values())
print("\nCurrent Setup:")
print(f"  - Total tenants: {len(TENANTS)}")
print(f"  - Total databases: {len(TENANTS)}")
print(f"  - Total collections: {total_collections}")
print(f"  - Average collections per tenant: {total_collections / len(TENANTS):.1f}")
Cleanup (Optional)
Uncomment and run the following cell to clean up the created databases.

# Cleanup - Uncomment to remove all tenant databases
# print("Cleaning up tenant databases...")
# for tenant_id, config in TENANTS.items():
#     try:
#         client.drop_database(config["db_name"])
#         print(f"✓ Dropped database {config['db_name']}")
#     except Exception as e:
#         print(f"Error dropping {config['db_name']}: {e}")
# print("Cleanup completed!")
Summary
This notebook demonstrated database-level multi-tenancy in Milvus, which provides:

Strongest Isolation: Complete data separation at the database level
Simple Management: Each tenant has their own namespace
Security: Zero risk of cross-tenant data access
Scalability Limit: Suitable for up to 64 enterprise tenants
This approach is ideal for enterprise customers requiring strict compliance and data isolation guarantees.

Chapter 17: Real-world Applications and Integrations — The book concludes by showing how Milvus integrates with frameworks like LangChain for RAG and image retrieval systems using pre-trained models like ResNet50.

Conclusion

I found this book incredibly useful. Even though, as mentioned, I consider myself a bit experienced with Milvus, the depth of technical detail helped me significantly strengthen my knowledge. Whether you are just starting with local testing or managing large-scale clusters on platforms like watsonx Data, this handbook provides the architectural clarity needed to build truly high-performance vector search systems.

>>> Thanks for reading <<<

Stop Treating Your Code Security Like a “Check Engine” Light (or part #2 of what we can do with Bob even if we’re not coders…)

Alain Airom (Ayrom) — Thu, 09 Apr 2026 12:13:01 +0000

Going to some of my old code and asking Bob to check for vulnerabilities!

Introduction

We’ve all been there. A little red light pops up on the dashboard, and instead of pulling over, we perform the ancient ritual of “The Great Ignore.” We tell ourselves it’s probably just a sensitive sensor or that the car feels fine, so it must be fine. We do the same thing with our digital infrastructure: we see a CVE notification, shrug, and decide that today isn’t the day to deal with a “theoretical” exploit.

But here’s the cold, hard reality: in the world of cybersecurity, that “Check Engine” light isn’t a glitch — it’s the sound of a digital crowbar prying at your front door.

While it’s easy to joke about the absurdity of a refrigerator getting hacked or a smart-toaster joining a botnet, the punchline stops being funny when it’s your customer data, your intellectual property, or your entire company’s reputation on the line. Security isn’t just a checkbox for the compliance department or a “nice-to-have” for when the budget allows; it’s the foundation that keeps the whole house from sliding down the hill.

In this post, I’m going to stop the nervous laughter and look at why treating vulnerabilities like minor inconveniences is the fastest way to become a cautionary tale. It’s time to take security seriously — before a hacker does it for you.

After the introduction, going to some of my old codes, I checked on an old code which I use by the way (locally) to merge my PDF files instead of going to a web site where I’m not quite sure about the privacy matters (by the way the code(s) are in an old blog post). So I asked Bob to to a security/vulnerability check. I got upset by myself from Bob findings 💀💀💀 😂.

As a refresher, I post again the shortest code which does not have a GUI (the Streamlit version is in the post)👇

# pdf_merger_2.py
import os
from PyPDF2 import PdfMerger, PdfReader

def merge_pdfs_in_directory(input_folder, output_folder, output_filename="merged_output.pdf"):
    """
    Recursively finds all PDF files in the input_folder and merges them
    into a single PDF file in the output_folder.

    Args:
        input_folder (str): The path to the folder containing PDF files.
        output_folder (str): The path to the folder where the merged PDF will be saved.
        output_filename (str): The name of the merged PDF file.
    """
    if not os.path.exists(input_folder):
        print(f"Error: Input folder '{input_folder}' does not exist.")
        return

    # --- This is the part that creates the output directory if it doesn't exist ---
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
        print(f"Created output folder: '{output_folder}'")
    # -----------------------------------------------------------------------------

    merger = PdfMerger()
    pdf_files_found = []

    # Walk through the input folder recursively
    for root, _, files in os.walk(input_folder):
        for file in files:
            if file.lower().endswith('.pdf'):
                filepath = os.path.join(root, file)
                pdf_files_found.append(filepath)

    if not pdf_files_found:
        print(f"No PDF files found in '{input_folder}' and its subdirectories.")
        return

    # Sort the files for consistent merging order (optional, but good practice)
    pdf_files_found.sort()

    print(f"Found {len(pdf_files_found)} PDF files to merge:")
    for pdf_file in pdf_files_found:
        print(f"- {pdf_file}")
        try:
            with open(pdf_file, 'rb') as f:
                merger.append(PdfReader(f))
        except Exception as e:
            print(f"Error appending {pdf_file}: {e}")

    output_filepath = os.path.join(output_folder, output_filename)

    try:
        with open(output_filepath, 'wb') as output_file:
            merger.write(output_file)
        print(f"\nSuccessfully merged PDF files to: '{output_filepath}'")
    except Exception as e:
        print(f"Error writing merged file: {e}")
    finally:
        merger.close()

if __name__ == "__main__":
    script_dir = os.path.dirname(os.path.abspath(__file__))
    input_folder_name = "input"
    output_folder_name = "output"

    # Construct full paths
    input_path = os.path.join(script_dir, input_folder_name)
    output_path = os.path.join(script_dir, output_folder_name)

    # Example Usage:
    # 1. Create an 'input' folder in the same directory as this script.
    # 2. Place some PDF files (and even subfolders with PDFs) inside the 'input' folder.
    # 3. Run this script.
    # The merged PDF will be created in the 'output' folder.
    merge_pdfs_in_directory(input_path, output_path, "all_merged_documents.pdf")

Bob’s anlysis was astonishing… 🫣

Lessons Learned and Conclusion

The security analysis of these two PDF merger applications serves as a stark reminder of how easily vulnerabilities can hide in even the most “straightforward” scripts. As we’ve seen, ignoring simple or complex security pitfalls — from uncontrolled resource consumption to path traversal — can transform a helpful tool into a significant liability.

While one might not “bother much” with extensive hardening for a quick local script, the standard changes entirely in a real-world software delivery environment. In professional production, these pitfalls must be double-checked with extreme rigor. Utilizing specialized tools like IBM Bob SDLC for security scanning and lifecycle management is an essential strategy for identifying these risks early; it ensures that even those who aren’t career developers can maintain a professional security posture and prevent their software from becoming a cautionary tale.

>>> Thanks for Reading <<<

My first Experience with Google’s Agent Development Kit (ADK-Go)

Alain Airom (Ayrom) — Tue, 07 Apr 2026 08:26:20 +0000

A first short experience with Google’s Go language Agent Development Kit (ADK-Go)

Introduction

The landscape of Artificial Intelligence is shifting from models that simply “chat” to agents that “do.” As developers look for more robust ways to integrate LLMs into real-world workflows, Google has released the Agent Development Kit (ADK) — an open-source framework designed to take the complexity out of building reliable, tool-augmented AI agents. Whether you are looking to automate multi-step research tasks, manage local file systems, or bridge the gap between cloud models and local execution, the ADK provides the standardized scaffolding necessary to move from experimental prompts to production-ready autonomous assistants. By open-sourcing this toolkit, Google is offering the developer community a structured, modular approach to session management, tool integration, and model orchestration, effectively lowering the barrier to entry for the next generation of “agentic” software.

Excerpt from ADK-Go Repository

_Agent Development Kit (ADK) is a flexible and modular framework that applies software development principles to AI agent creation. It is designed to simplify building, deploying, and orchestrating agent workflows, from simple tasks to complex systems. While optimized for Gemini, ADK is model-agnostic, deployment-agnostic, and compatible with other frameworks.

This Go version of ADK is ideal for developers building cloud-native agent applications, leveraging Go’s strengths in concurrency and performance.

Idiomatic Go: Designed to feel natural and leverage the power of Go.
Rich Tool Ecosystem: Utilize pre-built tools, custom functions, or integrate existing tools to give agents diverse capabilities.
Code-First Development: Define agent logic, tools, and orchestration directly in Go for ultimate flexibility, testability, and versioning.
Modular Multi-Agent Systems: Design scalable applications by composing multiple specialized agents.
Deploy Anywhere: Easily containerize and deploy agents, with strong support for cloud-native environments like Google Cloud Run._

Running the Provided Sample Out-of-the-Box!

Creating a basic agent is a matter of minutes using the example provided on the documentation site!

Obtain you Google API key which you should export as is or put in an .env file.

export GOOGLE_API_KEY="abcdefghixxxxxxxxxxxxx"

Copy/paste the code

package main

import (
    "context"
    "log"
    "os"

    "google.golang.org/adk/agent"
    "google.golang.org/adk/agent/llmagent"
    "google.golang.org/adk/cmd/launcher"
    "google.golang.org/adk/cmd/launcher/full"
    "google.golang.org/adk/model/gemini"
    "google.golang.org/adk/tool"
    "google.golang.org/adk/tool/geminitool"
    "google.golang.org/genai"
)

func main() {
    ctx := context.Background()

    model, err := gemini.NewModel(ctx, "gemini-2.5-flash", &genai.ClientConfig{
        APIKey: os.Getenv("GOOGLE_API_KEY"),
    })
    if err != nil {
        log.Fatalf("Failed to create model: %v", err)
    }

    timeAgent, err := llmagent.New(llmagent.Config{
        Name:        "hello_time_agent",
        Model:       model,
        Description: "Tells the current time in a specified city.",
        Instruction: "You are a helpful assistant that tells the current time in a city.",
        Tools: []tool.Tool{
            geminitool.GoogleSearch{},
        },
    })
    if err != nil {
        log.Fatalf("Failed to create agent: %v", err)
    }

    config := &launcher.Config{
        AgentLoader: agent.NewSingleLoader(timeAgent),
    }

    l := full.NewLauncher()
    if err = l.Execute(ctx, config, os.Args[1:]); err != nil {
        log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
    }
}

Configure the dependencies

go mod init my-agent/main
go mod tidy

You can run the application either in console mode or as a web app!

go run agent.go


User -> what is the time in Paris?

Agent -> The current time in Paris, France is 08:31 AM. Paris is currently observing Central European Summer Time (CEST), which is UTC+2.
User ->

ADK Web is not meant for use in production deployments. You should use ADK Web for development and debugging purposes only.

go run agent.go web api webui

Implementing A local Agent using Ollama and Lessons Learned

After a series of intensive integration tests exhausted my Google API quotas, I pivoted to a local development workflow. By leveraging Ollama to host models locally, I was able to continue refining the agent without API constraints. To speed up the transition, I brought in Bob, our AI construction specialist, to streamline the architecture and get the local environment running in record time.

First Test: Console Mode Application

The first attempt is just a simple Go application in console mode.

package main

import (
 "bufio"
 "context"
 "fmt"
 "iter"
 "log"
 "os"
 "os/exec"
 "strings"

 "google.golang.org/adk/agent"
 "google.golang.org/adk/agent/llmagent"
 "google.golang.org/adk/cmd/launcher"
 "google.golang.org/adk/cmd/launcher/full"
 "google.golang.org/adk/model"
 "google.golang.org/adk/session"
 "google.golang.org/adk/tool"
 "google.golang.org/adk/tool/geminitool"
 "google.golang.org/genai"

 openai "github.com/openai/openai-go/v3"
 "github.com/openai/openai-go/v3/option"
)

// OllamaModel wraps Ollama to implement the adk model.LLM interface
type OllamaModel struct {
 client    openai.Client
 modelName string
}

// NewOllamaModel creates a new Ollama model instance
func NewOllamaModel(modelName string, baseURL string) (*OllamaModel, error) {
 if baseURL == "" {
  baseURL = "http://localhost:11434/v1"
 }

 client := openai.NewClient(
  option.WithBaseURL(baseURL),
  option.WithAPIKey("ollama"), // Ollama doesn't need a real API key
 )

 return &OllamaModel{
  client:    client,
  modelName: modelName,
 }, nil
}

// Name returns the model name
func (o *OllamaModel) Name() string {
 return o.modelName
}

// GenerateContent implements the model.LLM interface
func (o *OllamaModel) GenerateContent(ctx context.Context, req *model.LLMRequest, stream bool) iter.Seq2[*model.LLMResponse, error] {
 return func(yield func(*model.LLMResponse, error) bool) {
  // Convert ADK request to OpenAI format
  messages := []openai.ChatCompletionMessageParamUnion{}

  // Add system instruction from config if present
  if req.Config != nil && req.Config.SystemInstruction != nil && len(req.Config.SystemInstruction.Parts) > 0 {
   var systemText strings.Builder
   for _, part := range req.Config.SystemInstruction.Parts {
    if part.Text != "" {
     systemText.WriteString(part.Text)
    }
   }
   if systemText.Len() > 0 {
    messages = append(messages, openai.SystemMessage(systemText.String()))
   }
  }

  // Add conversation history
  for _, content := range req.Contents {
   role := "user"
   if content.Role == "model" {
    role = "assistant"
   }

   var contentText strings.Builder
   for _, part := range content.Parts {
    if part.Text != "" {
     contentText.WriteString(part.Text)
    }
   }

   if contentText.Len() > 0 {
    if role == "user" {
     messages = append(messages, openai.UserMessage(contentText.String()))
    } else {
     messages = append(messages, openai.AssistantMessage(contentText.String()))
    }
   }
  }

  // Call Ollama via OpenAI-compatible API
  params := openai.ChatCompletionNewParams{
   Messages: messages,
   Model:    openai.ChatModel(o.modelName),
  }

  completion, err := o.client.Chat.Completions.New(ctx, params)
  if err != nil {
   yield(nil, fmt.Errorf("ollama API error: %w", err))
   return
  }

  // Convert response back to ADK format
  if len(completion.Choices) == 0 {
   yield(nil, fmt.Errorf("no response from Ollama"))
   return
  }

  responseText := completion.Choices[0].Message.Content

  response := &model.LLMResponse{
   Content: &genai.Content{
    Parts: []*genai.Part{
     genai.NewPartFromText(responseText),
    },
    Role: "model",
   },
  }

  yield(response, nil)
 }
}

func main() {
 // Load environment variables from .env file
 if data, err := os.ReadFile(".env"); err == nil {
  for _, line := range strings.Split(string(data), "\n") {
   line = strings.TrimSpace(line)
   if strings.HasPrefix(line, "export ") {
    line = strings.TrimPrefix(line, "export ")
   }
   if parts := strings.SplitN(line, "=", 2); len(parts) == 2 {
    key := strings.TrimSpace(parts[0])
    value := strings.Trim(strings.TrimSpace(parts[1]), "\"")
    os.Setenv(key, value)
   }
  }
 }

 // Set default model if not specified
 if os.Getenv("OLLAMA_MODEL") == "" {
  os.Setenv("OLLAMA_MODEL", "mistral:7b")
 }

 // Check if we are the sub-process worker
 if len(os.Args) > 1 && os.Args[1] == "internal-run" {
  runAgentWorker()
  return
 }

 // Interactive Loop
 fmt.Println("--- Ollama Agent (ADK) ---")
 fmt.Println("Type 'exit' to stop.")

 scanner := bufio.NewScanner(os.Stdin)
 for {
  fmt.Print("\nUser -> ")
  if !scanner.Scan() {
   break
  }

  input := strings.TrimSpace(scanner.Text())
  if input == "exit" || input == "quit" {
   fmt.Println("Goodbye!")
   break
  }
  if input == "" {
   continue
  }

  // Execute ourselves with the 'internal-run' flag
  cmd := exec.Command(os.Args[0], "internal-run", input)
  cmd.Stdout = os.Stdout
  cmd.Stderr = os.Stderr
  cmd.Env = os.Environ() // Pass environment variables to subprocess

  _ = cmd.Run()
 }
}

func runAgentWorker() {
 ctx := context.Background()

 // Get the user input from args
 if len(os.Args) < 3 {
  return
 }
 userInput := os.Args[2]

 // Create Ollama model - use model from environment or default
 modelName := os.Getenv("OLLAMA_MODEL")
 if modelName == "" {
  modelName = "mistral:7b"
 }

 ollamaModel, err := NewOllamaModel(modelName, "http://localhost:11434/v1")
 if err != nil {
  log.Fatalf("Failed to create Ollama model: %v", err)
 }
 log.Printf("DEBUG: Using Ollama model: %s", modelName)

 // Get Google API key for search tool
 apiKey := os.Getenv("GOOGLE_API_KEY")
 var tools []tool.Tool
 if apiKey != "" {
  // Add Google Search tool if API key is available
  tools = append(tools, geminitool.GoogleSearch{})
  log.Printf("DEBUG: Google Search tool added (API key present: %d chars)", len(apiKey))
 } else {
  log.Printf("DEBUG: No Google API key found, search tool not available")
 }

 // Create agent with Ollama model
 agentInstance, err := llmagent.New(llmagent.Config{
  Name:        "ollama_assistant",
  Model:       ollamaModel,
  Description: "A helpful assistant powered by IBM Granite via Ollama.",
  Instruction: "You are a helpful assistant. When asked about current information like time, weather, or recent events, you MUST use the Google Search tool to find accurate, up-to-date information. Do not make up answers or say you cannot access real-time data - use the search tool instead.",
  Tools:       tools,
 })
 if err != nil {
  log.Fatalf("Failed to create agent: %v", err)
 }

 // Create session service
 sessionService := session.InMemoryService()

 config := &launcher.Config{
  SessionService: sessionService,
  AgentLoader:    agent.NewSingleLoader(agentInstance),
 }

 l := full.NewLauncher()

 // Run console mode and pipe the user input to it
 // Create a pipe to send input
 r, w, err := os.Pipe()
 if err != nil {
  log.Fatalf("Failed to create pipe: %v", err)
 }
 oldStdin := os.Stdin
 os.Stdin = r

 // Write the input and exit command
 go func() {
  fmt.Fprintln(w, userInput)
  fmt.Fprintln(w, "exit")
  w.Close()
 }()

 // Execute in console mode
 if err := l.Execute(ctx, config, []string{"console"}); err != nil {
  os.Stdin = oldStdin
  log.Fatalf("Run failed: %v", err)
 }

 os.Stdin = oldStdin
}

// Made with Bob

The application runs as expected, even in console mode!

Lessons learned while making this application

What one should be aware of (either reading the documentation or learning by doing 😂);

The ADK is not designed to function as a s*tandalone CLI* or monolithic application. Instead, it is built as a highly modular agentic middleware. It is optimized to run in Interactive Mode, where it is orchestrated by other applications or agents, or in Web Mode, providing a robust backend for browser-based AI interfaces.
Another important fact: Not ALL models are fit for function calling!

While It is entirely possible to architect simple, zero-tool agents that function purely on the model’s pre-existing weights. While useful for general reasoning or creative tasks, these agents lack the real-time grounding provided by the ADK’s orchestration layer, making them less effective for data-sensitive or live environments.

# Ollama Model Configuration
# Models that likely support function calling (my list):
# - llama3.2:latest (recommended, 2GB)
# - llama3:latest (4.7GB)
# - mistral:7b (4.4GB)
# - deepseek-r1:latest (5.2GB)
#
# Models without function calling:
# - ibm/granite4:3b (default, no tools)
# - gemma3:4b (no tools)
# export OLLAMA_MODEL="llama3.2:latest"
export OLLAMA_MODEL="mistral:7b"

Second Test: Web UI Mode Application

Next, I wanted to see this agent in its natural habitat: a Web UI. While I prepped the environment, Bob essentially finished the job in two clicks. The result? A perfectly functional web interface that looks and feels exactly like the live demos on the ADK documentation site — Ollama integration and all.

package main

import (
 "context"
 "fmt"
 "iter"
 "log"
 "os"
 "strings"

 "google.golang.org/adk/agent"
 "google.golang.org/adk/agent/llmagent"
 "google.golang.org/adk/cmd/launcher"
 "google.golang.org/adk/cmd/launcher/full"
 "google.golang.org/adk/model"
 "google.golang.org/adk/tool"
 "google.golang.org/adk/tool/geminitool"
 "google.golang.org/genai"

 openai "github.com/openai/openai-go/v3"
 "github.com/openai/openai-go/v3/option"
)

// OllamaModel wraps Ollama to implement the adk model.LLM interface
type OllamaModel struct {
 client    openai.Client
 modelName string
}

// NewOllamaModel creates a new Ollama model instance
func NewOllamaModel(modelName string, baseURL string) (*OllamaModel, error) {
 if baseURL == "" {
  baseURL = "http://localhost:11434/v1"
 }

 client := openai.NewClient(
  option.WithBaseURL(baseURL),
  option.WithAPIKey("ollama"), // Ollama doesn't need a real API key
 )

 return &OllamaModel{
  client:    client,
  modelName: modelName,
 }, nil
}

// Name returns the model name
func (o *OllamaModel) Name() string {
 return o.modelName
}

// GenerateContent implements the model.LLM interface
func (o *OllamaModel) GenerateContent(ctx context.Context, req *model.LLMRequest, stream bool) iter.Seq2[*model.LLMResponse, error] {
 return func(yield func(*model.LLMResponse, error) bool) {
  // Convert ADK request to OpenAI format
  messages := []openai.ChatCompletionMessageParamUnion{}

  // Add system instruction from config if present
  if req.Config != nil && req.Config.SystemInstruction != nil && len(req.Config.SystemInstruction.Parts) > 0 {
   var systemText strings.Builder
   for _, part := range req.Config.SystemInstruction.Parts {
    if part.Text != "" {
     systemText.WriteString(part.Text)
    }
   }
   if systemText.Len() > 0 {
    messages = append(messages, openai.SystemMessage(systemText.String()))
   }
  }

  // Add conversation history
  for _, content := range req.Contents {
   role := "user"
   if content.Role == "model" {
    role = "assistant"
   }

   var contentText strings.Builder
   for _, part := range content.Parts {
    if part.Text != "" {
     contentText.WriteString(part.Text)
    }
   }

   if contentText.Len() > 0 {
    if role == "user" {
     messages = append(messages, openai.UserMessage(contentText.String()))
    } else {
     messages = append(messages, openai.AssistantMessage(contentText.String()))
    }
   }
  }

  // Call Ollama via OpenAI-compatible API
  params := openai.ChatCompletionNewParams{
   Messages: messages,
   Model:    openai.ChatModel(o.modelName),
  }

  completion, err := o.client.Chat.Completions.New(ctx, params)
  if err != nil {
   yield(nil, fmt.Errorf("ollama API error: %w", err))
   return
  }

  // Convert response back to ADK format
  if len(completion.Choices) == 0 {
   yield(nil, fmt.Errorf("no response from Ollama"))
   return
  }

  responseText := completion.Choices[0].Message.Content

  response := &model.LLMResponse{
   Content: &genai.Content{
    Parts: []*genai.Part{
     genai.NewPartFromText(responseText),
    },
    Role: "model",
   },
  }

  yield(response, nil)
 }
}

func main() {
 // Load environment variables from .env file
 if data, err := os.ReadFile(".env"); err == nil {
  for _, line := range strings.Split(string(data), "\n") {
   line = strings.TrimSpace(line)
   if strings.HasPrefix(line, "export ") {
    line = strings.TrimPrefix(line, "export ")
   }
   if parts := strings.SplitN(line, "=", 2); len(parts) == 2 {
    key := strings.TrimSpace(parts[0])
    value := strings.Trim(strings.TrimSpace(parts[1]), "\"")
    os.Setenv(key, value)
   }
  }
 }

 ctx := context.Background()

 // Set default model if not specified
 modelName := os.Getenv("OLLAMA_MODEL")
 if modelName == "" {
  modelName = "mistral:7b"
 }

 // Create Ollama model
 ollamaModel, err := NewOllamaModel(modelName, "http://localhost:11434/v1")
 if err != nil {
  log.Fatalf("Failed to create Ollama model: %v", err)
 }
 log.Printf("Using Ollama model: %s", modelName)

 // Get Google API key for search tool
 apiKey := os.Getenv("GOOGLE_API_KEY")
 var tools []tool.Tool
 if apiKey != "" {
  // Add Google Search tool if API key is available
  tools = append(tools, geminitool.GoogleSearch{})
  log.Printf("Google Search tool enabled")
 } else {
  log.Printf("No Google API key found, search tool not available")
 }

 // Create agent with Ollama model
 agentInstance, err := llmagent.New(llmagent.Config{
  Name:        "ollama_assistant",
  Model:       ollamaModel,
  Description: "A helpful assistant powered via Ollama and ADK.",
  Instruction: "You are a helpful assistant. When asked about current information like time, weather, or recent events, you MUST use the Google Search tool to find accurate, up-to-date information. Do not make up answers or say you cannot access real-time data - use the search tool instead.",
  Tools:       tools,
 })
 if err != nil {
  log.Fatalf("Failed to create agent: %v", err)
 }

 config := &launcher.Config{
  AgentLoader: agent.NewSingleLoader(agentInstance),
 }

 l := full.NewLauncher()

 // Execute with command line arguments (supports web api webui)
 if err = l.Execute(ctx, config, os.Args[1:]); err != nil {
  log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
 }
}

// Made with Bob

Conclusion

The Agent Development Kit (ADK) is a framework that prioritizes interaction over isolation, designed to transform static LLMs into dynamic collaborators that can reach out to the world via Google Search or run privately through Ollama. By moving the logic into a Web UI and leveraging Bob’s structural guidance, we can demonstrate that the ADK isn’t just a tool for building chatbots — it’s an engine for creating “agentic” workflows that are as at home in a browser as they are in a complex multi-agent ecosystem. Whether you’re relying on a model’s internal knowledge or empowering it with real-time data, the ADK provides the standardized bridge needed to turn AI potential into practical, integrated reality.

>>> Thanks for reading <<<

Things you can do with IBM Bob if you don’t want to Code!

Alain Airom (Ayrom) — Sun, 05 Apr 2026 11:55:31 +0000

Using IBM Bob SDLC in documentation mode.

Introduction

In the fast-paced world of software development, it is easy to pigeonhole an AI assistant like IBM Bob SDLC as just another code-crunching engine. While its ability to churn out clean, efficient syntax is undeniable, its true value lies in its versatility across the entire project lifecycle. Whether you are a junior developer looking to bridge the gap between logic and implementation, or a Product Owner, Functional Consultant, or Project Manager navigating the complexities of requirements, Bob adapts to your needs. From drafting comprehensive technical specifications to performing the “detective work” of retro-documentation on legacy systems, IBM Bob is less of a simple tool and more of a multi-faceted collaborator designed to streamline workflows for every stakeholder involved.

To illustrate this, let’s jump into a hypothetical case of retro documentation of a whole application based on a code repository (a GitHub repository). I asked candidly from Bob to provide a detailed an architecture documentation for the IBM Instana Go-Sensor code! And Bob did it for sure 🤓 (even though the code is well documented), but my aim is to showcase one of the document writing capacities of Bob!

TL;DR-What is Instana?

Instana STAN

At its core, IBM Instana is an enterprise-grade observability and Application Performance Monitoring (APM) platform designed for the complexity of modern, cloud-native environments. Unlike traditional monitoring tools that require manual configuration and struggle with the ephemeral nature of microservices, Instana is built on automation. It utilizes “Agentic AI” and over 250 domain-specific sensors to automatically discover, map, and monitor every component of your technology stack — from mobile front-ends and web browsers down to the underlying infrastructure, including Kubernetes, serverless functions, and even legacy mainframes.

It captures high-fidelity data with one-second granularity and traces every single request without sampling, ensuring that no performance anomaly or error goes unnoticed. By leveraging AI-powered root-cause analysis, Instana doesn’t just alert you that something is wrong; it correlates events to tell you why it happened, drastically reducing the Mean Time to Repair (MTTR). This “democratization of data” allows DevOps, SREs, and Product Owners to move away from reactive “war rooms” and toward proactive optimization, ensuring a seamless and reliable end-user experience across the entire digital journey.

When Bob meets Stan 😂

OK, having introduced Instana, for my test usecase, I gave the following prompt to Bob, and then I show the interactions between Bob and myself!

Read the following Github: https://github.com/instana/go-sensor and provide a full arhitectural document of the repository and waht the code does alongside with mermaid flow charts 
I want the architecture mermaid flow charts added as well

Then a whole task list is provided, the user can always stop Bob and give further instructions;

Once I validated the tasks (or if you authorize Bob to do specific tasks without approval for all steps…), I got a full architecture document in Markdown format, with revelant flows and a detailed documentation. The one shown below is the high-level architecture schema.

There are also several specific area flows provided.

The document contains also description of components and in this case a recap table of “Framework Instrumentation Packages”.

Framework Instrumentation Packages

| Package              | Framework        | Features                    |
| -------------------- | ---------------- | --------------------------- |
| `instagin`           | Gin              | Middleware for HTTP routing |
| `instaecho`          | Echo             | HTTP middleware             |
| `instafiber`         | Fiber            | HTTP middleware             |
| `instamux`           | Gorilla Mux      | HTTP middleware             |
| `instabeego`         | Beego            | HTTP middleware             |
| `instafasthttp`      | FastHTTP         | HTTP middleware             |
| `instagrpc`          | gRPC             | Unary & stream interceptors |
| `instasarama`        | Kafka (Sarama)   | Producer/consumer wrappers  |
| `instaamqp091`       | RabbitMQ         | Publisher/consumer wrappers |
| `instamongo`         | MongoDB          | Command monitoring          |
| `instaredis`         | Redis (go-redis) | Command hooks               |
| `instaredigo`        | Redis (Redigo)   | Connection wrapper          |
| `instagorm`          | GORM             | Callbacks for DB operations |
| `instagocb`          | Couchbase        | Bucket wrapper              |
| `instapgx`           | PostgreSQL (pgx) | Tracer implementation       |
| `instacosmos`        | Azure Cosmos DB  | Client wrapper              |
| `instaawssdk`        | AWS SDK v1       | Service handlers            |
| `instaawsv2`         | AWS SDK v2       | Middleware                  |
| `instalambda`        | AWS Lambda       | Handler wrapper             |
| `instaazurefunction` | Azure Functions  | Handler wrapper             |
| `instagraphql`       | GraphQL          | Resolver middleware         |

Bob can also provide a documentation on specific part or parts of an existing code. Let’s imagine that hypothetically we have an undocumented code or a code in a programming language we do not know!

// (c) Copyright IBM Corp. 2021
// (c) Copyright Instana Inc. 2020

package instana

import (
 "bytes"
 "context"
 "encoding/json"
 "fmt"
 "io"
 "net/http"
 "os"
 "strconv"
 "sync"
 "time"

 "github.com/instana/go-sensor/acceptor"
 "github.com/instana/go-sensor/autoprofile"
)

const awsLambdaAgentFlushPeriod = 2 * time.Second

type lambdaAgent struct {
 Endpoint string
 Key      string
 PID      int

 snapshot serverlessSnapshot

 mu        sync.RWMutex
 spanQueue []Span

 client *http.Client
 logger LeveledLogger
}

func newLambdaAgent(
 serviceName, acceptorEndpoint, agentKey string,
 client *http.Client,
 logger LeveledLogger,
) *lambdaAgent {
 if logger == nil {
  logger = defaultLogger
 }

 if client == nil {
  client = http.DefaultClient
 }

 logger.Debug("initializing aws lambda agent")

 agent := &lambdaAgent{
  Endpoint: acceptorEndpoint,
  Key:      agentKey,
  PID:      os.Getpid(),
  client:   client,
  logger:   logger,
 }

 go func(a *lambdaAgent) {
  t := time.NewTicker(awsLambdaAgentFlushPeriod)
  defer t.Stop()

  for range t.C {
   if err := a.Flush(context.Background()); err != nil {
    a.logger.Error("failed to post collected data: ", err)
   }
  }
 }(agent)

 return agent
}

func (a *lambdaAgent) Ready() bool { return true }

func (a *lambdaAgent) SendMetrics(data acceptor.Metrics) error { return nil }

func (a *lambdaAgent) SendEvent(event *EventData) error { return nil }

func (a *lambdaAgent) SendSpans(spans []Span) error {
 a.enqueueSpans(spans)
 return nil
}

func (a *lambdaAgent) SendProfiles(profiles []autoprofile.Profile) error { return nil }

func (a *lambdaAgent) Flush(ctx context.Context) error {
 snapshot := a.collectSnapshot(a.spanQueue)

 if snapshot.EntityID == "" {
  return ErrAgentNotReady
 }

 from := newServerlessAgentFromS(snapshot.EntityID, "aws")

 payload := struct {
  Metrics metricsPayload `json:"metrics,omitempty"`
  Spans   []Span         `json:"spans,omitempty"`
 }{
  Metrics: metricsPayload{
   Plugins: []acceptor.PluginPayload{
    acceptor.NewAWSLambdaPluginPayload(snapshot.EntityID),
   },
  },
 }

 a.mu.Lock()
 payload.Spans = make([]Span, len(a.spanQueue))
 copy(payload.Spans, a.spanQueue)
 a.spanQueue = a.spanQueue[:0]
 a.mu.Unlock()

 for i := range payload.Spans {
  payload.Spans[i].From = from
 }

 buf := bytes.NewBuffer(nil)
 if err := json.NewEncoder(buf).Encode(payload); err != nil {
  return fmt.Errorf("failed to marshal traces payload: %s", err)
 }

 req, err := http.NewRequest(http.MethodPost, a.Endpoint+"/bundle", buf)
 if err != nil {
  a.enqueueSpans(payload.Spans)
  return fmt.Errorf("failed to prepare send traces request: %s", err)
 }

 req.Header.Set("Content-Type", "application/json")

 if err := a.sendRequest(req.WithContext(ctx)); err != nil {
  a.enqueueSpans(payload.Spans)
  return fmt.Errorf("failed to send traces, will retry later: %s", err)
 }

 return nil
}

func (a *lambdaAgent) enqueueSpans(spans []Span) {
 a.mu.Lock()
 defer a.mu.Unlock()

 a.spanQueue = append(a.spanQueue, spans...)
}

func (a *lambdaAgent) sendRequest(req *http.Request) error {
 a.mu.RLock()
 host := a.snapshot.Host
 a.mu.RUnlock()

 req.Header.Set("X-Instana-Host", host)
 req.Header.Set("X-Instana-Key", a.Key)
 req.Header.Set("X-Instana-Time", strconv.FormatInt(time.Now().UnixNano()/int64(time.Millisecond), 10))

 resp, err := a.client.Do(req)
 if err != nil {
  return fmt.Errorf("failed to send request to the serverless agent: %s", err)
 }

 defer resp.Body.Close()

 if resp.StatusCode >= http.StatusBadRequest {
  respBody, err := io.ReadAll(resp.Body)
  if err != nil {
   a.logger.Debug("failed to read serverless agent response: ", err)
   return nil
  }

  a.logger.Info("serverless agent has responded with ", resp.Status, ": ", string(respBody))
  return nil
 }

 io.CopyN(io.Discard, resp.Body, 1<<20)

 return nil
}

func (a *lambdaAgent) collectSnapshot(spans []Span) serverlessSnapshot {
 a.mu.RLock()
 entityID := a.snapshot.EntityID
 a.mu.RUnlock()

 if entityID != "" {
  return a.snapshot
 }

 // searching for the lambda entry span in reverse order, since it's
 // more likely to be finished last
 for i := len(spans) - 1; i >= 0; i-- {
  sp, ok := spans[i].Data.(AWSLambdaSpanData)
  if !ok {
   continue
  }

  a.mu.Lock()
  a.snapshot = serverlessSnapshot{
   EntityID: sp.Snapshot.ARN,
   Host:     sp.Snapshot.ARN,
  }
  a.mu.Unlock()
  a.logger.Debug("collected snapshot")

  break
 }

 return a.snapshot
}

I gave the following prompt to Bob ⤵️

Provide a document explainig the "lambda_agent.go" functionality, what it does and how it can be used

Not only Bob provides the very detailed documentation, but also a full detailed component architecture of tho code! I provide hereafter a portion of the document generated.

-Core Components

The Lambda agent consists of several key components:

1. **Span Queue**: Thread-safe buffer for collecting trace spans
2. **Snapshot Collector**: Captures Lambda function metadata (ARN, version, etc.)
3. **Periodic Flusher**: Automatically sends data every 2 seconds
4. **HTTP Client**: Handles communication with Instana backend

-Key Features

-1. Automatic Span Collection

The agent automatically collects and queues trace spans generated during Lambda execution:

- **Thread-safe queuing**: Uses mutex locks to prevent race conditions
- **Automatic batching**: Groups spans for efficient transmission
- **Retry mechanism**: Re-queues spans if sending fails

-2. Serverless Snapshot

Captures essential Lambda function metadata:

- **ARN (Amazon Resource Name)**: Unique identifier for the Lambda function
- **Host information**: Lambda execution environment details
- **Process ID**: Current Lambda container process

-3. Periodic Flushing

Automatically sends collected data at regular intervals:

- **Flush period**: 2 seconds (configurable via `awsLambdaAgentFlushPeriod`)
- **Background goroutine**: Non-blocking operation
- **Context-aware**: Respects cancellation and timeouts

-4. Metrics Integration

Includes AWS Lambda plugin payload with each flush:

- **Entity identification**: Links spans to Lambda function
- **Cloud provider metadata**: AWS-specific information
- **Performance metrics**: Execution time, cold starts, etc.

-Snapshot Collection

The agent extracts Lambda metadata from the first `aws.lambda.entry` span:

1. Searches span queue in reverse order (most recent first)
2. Finds span with `AWSLambdaSpanData` type
3. Extracts ARN from span snapshot
4. Caches snapshot for subsequent flushes

-Configuration

-Environment Variables

The Lambda agent is automatically configured when running in AWS Lambda environment:

| Variable               | Description                        | Required | Default |
| ---------------------- | ---------------------------------- | -------- | ------- |
| `INSTANA_AGENT_KEY`    | Instana backend authentication key | Yes      | -       |
| `INSTANA_ENDPOINT_URL` | Instana backend endpoint           | Yes      | -       |
| `AWS_EXECUTION_ENV`    | AWS Lambda execution environment   | Auto-set | -       |
| `INSTANA_ZONE`         | Availability zone tag              | No       | -       |
| `INSTANA_TAGS`         | Custom tags (comma-separated)      | No       | -       |
| `INSTANA_SECRETS`      | Secret patterns to redact          | No       | -       |

Conclusion

Ultimately, the integration of IBM Bob SDLC into the analysis of a public repository demonstrates that its value extends far beyond the IDE. By processing the raw complexity of an existing codebase, Bob acts as a bridge between technical debt and clarity. In this specific scenario, its ability to generate retro-documentation — essentially reverse-engineering the “why” behind the “what” — allows teams to reclaim lost knowledge without manual archeology. When paired with application architecture mapping and granular code documentation, Bob transforms a dense thicket of files into a transparent, navigable roadmap.

This synergy proves that Bob is an essential asset for more than just the engineering team. For the Product Owner or Project Manager, these documentation capabilities provide a high-level view of system dependencies and health; for the Functional Consultant, they offer a clear understanding of logic flows. By automating the most tedious aspects of the SDLC, Bob ensures that whether you are onboarding a new junior developer or planning a massive architectural migration, the “source of truth” is always documented, up-to-date, and accessible to everyone involved.

>>> Thanks for reading <<<

Granite 🪨 4.0 3B 👓 Vision: Compact Multimodal Intelligence for Enterprise Documents is out and Bob built a UI to use it!

Alain Airom (Ayrom) — Wed, 01 Apr 2026 15:48:16 +0000

Creating a UI for the latest Granite Vision Model (using Docling)!

Introduction

IBM just annonced on Hugging Face the availability for “Granite 4.0 3b vision” model. This is a vision-language model (VLM) designed for enterprise-grade document data extraction. It focuses on specialized, complex extraction tasks that ultracompact models often struggle with:

Chart extraction: Converting charts into structured, machine-readable formats (Chart2CSV, Chart2Summary, and Chart2Code),
Table extraction: Accurately extracting tables with complex layouts from document images to JSON, HTML, or OTSL,
Semantic Key-Value Pair (KVP) extraction: Extracting values based on key names and descriptions across diverse document layouts.

Supported Tasks

The model supports specialized extraction tasks, each activated by a simple task tag in the user message. The chat template automatically expands tags into the full prompt — no need to write verbose instructions.


|                 Tag                 |                  Task                   |                  Output                   |
| :---------------------------------: | :-------------------------------------: | :---------------------------------------: |
|            `<chart2csv>`            |              Chart to CSV               | CSV table with headers and numeric values |
|           `<chart2code>`            |          Chart to Python code           |   Python code that recreates the chart    |
|          `<chart2summary>`          |            Chart to summary             | Natural-language description of the chart |
|           `<tables_json>`           |         Table extraction (JSON)         | Structured JSON with dimensions and cells |
|           `<tables_html>`           |         Table extraction (HTML)         |           HTML `<table>` markup           |
|           `<tables_otsl>`           |         Table extraction (OTSL)         |     OTSL markup with cell/merge tags      |
| KVP (see prompt instructions below) | Schema based Key-Value pairs extraction | JSON with nested dictionaries and arrays  |

OK, having said that let’s see how you can implement, test and use it!

Implementation

Using the code provided on Hugging Face

You can use the code provided on Hugging Face page of the model (link provided in Links section) and run it!

Prepare your virtual environment and install the required packages

pip install torch==2.10.0 --index-url https://download.pytorch.org/whl/cu128
pip install transformers==4.57.6 peft==0.18.1 tokenizers==0.22.2 pillow==12.1.1

Usage with transformers;

import re
from io import StringIO

import pandas as pd
import torch
from transformers import AutoProcessor, AutoModelForImageTextToText
from PIL import Image
from huggingface_hub import hf_hub_download

model_id = "ibm-granite/granite-4.0-3b-vision"
device = "cuda" if torch.cuda.is_available() else "cpu"

processor = AutoProcessor.from_pretrained(model_id, trust_remote_code=True)
model = AutoModelForImageTextToText.from_pretrained(
    model_id,
    trust_remote_code=True,
    dtype=torch.bfloat16,
    device_map=device
).eval()

# Optional: merge LoRA adapters into base weights for faster inference.
# Prefer to skip when using text-only tasks, as the LoRA adapters are vision-specific.
model.merge_lora_adapters()

def run_inference(model, processor, images, prompts):
    """Run batched inference on image+prompt pairs (one image per prompt)."""
    conversations = [
        [{"role": "user", "content": [
            {"type": "image"},
            {"type": "text", "text": prompt},
        ]}]
        for prompt in prompts
    ]
    texts = [
        processor.apply_chat_template(conv, tokenize=False, add_generation_prompt=True)
        for conv in conversations
    ]
    inputs = processor(
        text=texts, images=images, return_tensors="pt", padding=True, do_pad=True
    ).to(model.device)
    outputs = model.generate(
        **inputs,
        max_new_tokens=4096, 
        use_cache=True
    )
    results = []
    for i in range(len(prompts)):
        gen = outputs[i, inputs["input_ids"].shape[1]:]
        results.append(processor.decode(gen, skip_special_tokens=True))
    return results


def display_table(text):
    """Pretty-print CSV (possibly wrapped in ```

csv

```) or HTML table content via pandas."""
    m = re.search(r"```

csv\s*
(.*?)

```", text, re.DOTALL)
    if m:
        df = pd.read_csv(StringIO(m.group(1)))
        print(df.to_string(index=False))
    elif "<table" in text.lower():
        df = pd.read_html(StringIO(text))[0]
        print(df.to_string(index=False))
    else:
        print(text)

Charts and Tables;

chart_path = hf_hub_download(repo_id=model_id, filename="chart.jpg")
table_path = hf_hub_download(repo_id=model_id, filename="table.png")
chart_img = Image.open(chart_path).convert("RGB")
table_img = Image.open(table_path).convert("RGB")

# Batched chart tasks
chart_prompts = ["<chart2csv>", "<chart2summary>", "<chart2code>"]
chart_results = run_inference(model, processor, [chart_img] * len(chart_prompts), chart_prompts)
for prompt, result in zip(chart_prompts, chart_results):
    print(f"{prompt}:")
    display_table(result)
    print()

# Batched table tasks
table_prompts = ["<tables_html>", "<tables_otsl>"]
table_results = run_inference(model, processor, [table_img] * len(table_prompts), table_prompts)
for prompt, result in zip(table_prompts, table_results):
    print(f"{prompt}:")
    display_table(result)
    print()

KVP — Key Value Extraction

import json

invoice_path = hf_hub_download(repo_id=model_id, filename="invoice.png")
invoice_img = Image.open(invoice_path).convert("RGB")
schema = {
    "type": "object",
    "properties": {
        "invoice_date": {"type": "string", "description": "The date the invoice was issued"},
        "order_number": {"type": "string", "description": "The unique identifier for the order"},
        "seller_tax_id": {"type": "string", "description": "The tax identification number of the seller"},
    }
}

prompt = f"""Extract structured data from this document.
Return a JSON object matching this schema:

{json.dumps(schema, indent=2)}

Return null for fields you cannot find.
Return ONLY valid JSON.
Return an instance of the JSON with extracted values, not the schema itself."""

result = run_inference(model, processor, [invoice_img], [prompt])[0]
print(result)

I made two simple applications implementing the Granite model, one built with Streamlit and a second one (enhance UI, more user-friendly, using IBM Bob).

Streamlit version including Docling (no Bob)

Addtitional packages are needed for this version.

Install the requirements (I had many version conflicts so you should adapt to your configuration)

beautifulsoup4>=4.12.0
lxml>=4.9.0
markdown>=3.5.0
python-dotenv>=1.0.0

docling>=2.0.0
pdf2image>=1.16.3

torch==2.10.0 --index-url https://download.pytorch.org/whl/cu128
transformers==4.57.6 
peft==0.18.1 
tokenizers==0.22.2 
pillow==12.1.1 
streamlit 
pandas

And the application

import streamlit as st
import torch
from transformers import AutoProcessor, AutoModel 
from PIL import Image
import os
import traceback
from datetime import datetime

# --- DOCLING IMPORTS ---
from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.pipeline_options import PdfPipelineOptions
from docling.datamodel.base_models import InputFormat

# --- CONFIGURATION ---
MODEL_ID = "ibm-granite/granite-4.0-3b-vision"
DEVICE = "mps" if torch.backends.mps.is_available() else "cpu"
OUTPUT_DIR = "output"

if not os.path.exists(OUTPUT_DIR):
    os.makedirs(OUTPUT_DIR)

st.set_page_config(page_title="Granite 4.0 Vision + Docling v4", layout="wide")

@st.cache_resource
def load_tools():
    try:
        with st.spinner("Step 1: Loading IBM Granite 4.0 Vision..."):
            processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
            model = AutoModel.from_pretrained(MODEL_ID, torch_dtype=torch.bfloat16, device_map=DEVICE, trust_remote_code=True)
            if hasattr(model, "merge_lora_adapters"):
                model.merge_lora_adapters()
        with st.spinner("Step 2: Initializing Docling Engine..."):
            pipeline_options = PdfPipelineOptions()
            pipeline_options.generate_page_images = True
            pipeline_options.images_scale = 2.0 
            pdf_options = PdfFormatOption(pipeline_options=pipeline_options)
            converter = DocumentConverter(format_options={InputFormat.PDF: pdf_options})
        return processor, model, converter
    except Exception as e:
        st.error(f"Initialization Error: {e}")
        return None, None, None

processor, model, converter = load_tools()

def run_vlm(image, task_prompt, _processor, _model):
    prompt = f"<|user|>\n<image>\n{task_prompt}<|assistant|>\n"
    inputs = _processor(images=image, text=prompt, return_tensors="pt").to(DEVICE, torch.bfloat16)
    output_tokens = _model.generate(**inputs, max_new_tokens=1024, do_sample=False, num_beams=1)
    return _processor.batch_decode(output_tokens, skip_special_tokens=True)[0].strip()

st.title("📄 Granite 4.0 Vision + Docling v3")
if processor is None or converter is None:
    st.stop()

TASK_MAP = {
    "Table (HTML)": {"prompt": "<table_html>", "ext": "html"},
    "Table (JSON)": {"prompt": "<table_json>", "ext": "json"},
    "Key-Value Pairs (KVP)": {"prompt": "<kvp>", "ext": "json"},
    "Chart (CSV)": {"prompt": "<chart_csv>", "ext": "csv"}
}

with st.sidebar:
    st.header("⚙️ Settings")
    do_markdown = st.checkbox("Export Full Markdown (Docling Native)", value=True)
    selected_tasks = st.multiselect("Granite AI Tasks:", list(TASK_MAP.keys()), default=["Table (HTML)"])
    if st.button("Clear Output Folder"):
        for f in os.listdir(OUTPUT_DIR): os.remove(os.path.join(OUTPUT_DIR, f))

uploaded_file = st.file_uploader("Upload PDF", type=["pdf", "png", "jpg", "jpeg"])

if uploaded_file:
    temp_path = os.path.join(os.getcwd(), f"temp_{uploaded_file.name}")
    with open(temp_path, "wb") as f: f.write(uploaded_file.getbuffer())

    if st.button("🚀 Start Extraction"):
        try:
            with st.status("Processing Document...") as status:
                conv_result = converter.convert(temp_path)

                if do_markdown:
                    md_output = conv_result.document.export_to_markdown()
                    with open(os.path.join(OUTPUT_DIR, f"doc_{datetime.now().strftime('%H%M%S')}.md"), "w") as f:
                        f.write(md_output)
                    st.success("Markdown Export Complete")
                    with st.expander("📝 View Markdown"): st.markdown(md_output)

                for page_no, page_obj in conv_result.document.pages.items():
                    st.subheader(f"📍 Page {page_no}")
                    if page_obj.image:
                        # Fix: Use .pil_image for ImageRef compatibility
                        page_img = page_obj.image.pil_image 
                        st.image(page_img, width=450)
                        for task in selected_tasks:
                            with st.spinner(f"Running {task}..."):
                                res = run_vlm(page_img, TASK_MAP[task]["prompt"], processor, model)
                                out_name = f"pg{page_no}_{task.replace(' ', '_')}_{datetime.now().strftime('%H%M%S')}.{TASK_MAP[task]['ext']}"
                                with open(os.path.join(OUTPUT_DIR, out_name), "w") as f: f.write(res)
                                with st.expander(f"✅ {task}"): st.code(res)
            st.success("All Tasks Finished!")
        except Exception as e:
            st.error(f"Runtime Error: {e}")
            print(traceback.format_exc())
        finally:
            if os.path.exists(temp_path): os.remove(temp_path)

I used images and real invoices, the output for a real invoice from my medical visit in French, from PDF too Markdown) is quite outstanding!

-Facture n° 1775 du 01/04/2026

Feuille de soins électronique

- xxxx · EI

Masseur Kinésithérapeute

xxxxx N° facturation PS : xxxxxx RPPS : 11111111

- Bénéficiaire des soins

YYYY Né le xx/xx/xxxx NIR : 1234567890 NANTERRE (xxx) Régime Général (xx)

-Assuré

YYYY

Né le 01/01/1900

NIR : 1234567890

| Date       | Exonération | Acte     | Taux AMO | Part AMO | Part AMC | Montant |
| ---------- | ----------- | -------- | -------- | -------- | -------- | ------- |
| 25/03/2026 | -           | VSM 8.08 | 60,00 %  | 10,72 €  | 0,00 €   | 38,00 € |
| 01/04/2026 | -           | VSM 8.08 | 60,00 %  | 10,72 €  | 0,00 €   | 38,00 € |

Payé le 01/04/2026 par Carte bancaire

| TOTAL        | 76,00 € |
| ------------ | ------- |
| Part patient | 54,56 € |
| Part AMO     | 21,44 € |
| Part AMC     | 0,00 €  |

<!-- image -->

Prescripteur N°12345 Le 09/03/2026

Bob’s version built on Gradio including Docling

To create the version which is presented below, I used Bob, and gave to it the Streamlit built version. I asked for a Gradio based version and an enhanced UI.

This is the UI of what Bob provided;

And the following is the code;

#!/usr/bin/env python3
"""
Enhanced Gradio UI for Granite 4.0 Vision + Docling
Features:
- Better UI with live previews
- Multi-choice checkboxes for output formats
- Page-by-page processing with previews
- Real-time result display
"""

import gradio as gr
import torch
from transformers import AutoProcessor, AutoModel
from PIL import Image
import os
import traceback
from datetime import datetime
from pathlib import Path
import json

# Docling imports
from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.pipeline_options import PdfPipelineOptions
from docling.datamodel.base_models import InputFormat

# Configuration
MODEL_ID = "ibm-granite/granite-4.0-3b-vision"
DEVICE = "mps" if torch.backends.mps.is_available() else "cpu"
OUTPUT_DIR = "output"

# Ensure output directory exists
Path(OUTPUT_DIR).mkdir(exist_ok=True)

# Task definitions
TASK_MAP = {
    "Table (HTML)": {"prompt": "<table_html>", "ext": "html", "icon": "📊"},
    "Table (JSON)": {"prompt": "<table_json>", "ext": "json", "icon": "📋"},
    "Key-Value Pairs (KVP)": {"prompt": "<kvp>", "ext": "json", "icon": "🔑"},
    "Chart (CSV)": {"prompt": "<chart_csv>", "ext": "csv", "icon": "📈"},
}

# Global variables for model and converter
processor = None
model = None
converter = None


def initialize_models():
    """Initialize Granite Vision model and Docling converter."""
    global processor, model, converter

    try:
        # Load Granite Vision model
        print(f"Loading Granite 4.0 Vision on {DEVICE}...")
        processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
        model = AutoModel.from_pretrained(
            MODEL_ID,
            torch_dtype=torch.bfloat16,
            device_map=DEVICE,
            trust_remote_code=True
        )

        if hasattr(model, "merge_lora_adapters"):
            model.merge_lora_adapters()

        # Initialize Docling converter
        print("Initializing Docling converter...")
        pipeline_options = PdfPipelineOptions()
        pipeline_options.generate_page_images = True
        pipeline_options.images_scale = 2.0
        pdf_options = PdfFormatOption(pipeline_options=pipeline_options)
        converter = DocumentConverter(format_options={InputFormat.PDF: pdf_options})

        return "✅ Models initialized successfully!"

    except Exception as e:
        return f"❌ Error initializing models: {str(e)}\n{traceback.format_exc()}"


def run_vlm_task(image, task_prompt):
    """Run a VLM task on an image."""
    if processor is None or model is None:
        return "Error: Models not initialized"

    try:
        prompt = f"<|user|>\n<image>\n{task_prompt}<|assistant|>\n"
        inputs = processor(images=image, text=prompt, return_tensors="pt").to(DEVICE, torch.bfloat16)
        output_tokens = model.generate(**inputs, max_new_tokens=1024, do_sample=False, num_beams=1)
        result = processor.batch_decode(output_tokens, skip_special_tokens=True)[0].strip()
        return result
    except Exception as e:
        return f"Error: {str(e)}"


def format_preview(content, format_type):
    """Format content for preview based on type."""
    if format_type == "html":
        return content
    elif format_type in ["json", "csv"]:
        return f"```
{% endraw %}
{format_type}\n{content}\n
{% raw %}
```"
    else:
        return content


def process_document(
    file,
    export_markdown,
    task_table_html,
    task_table_json,
    task_kvp,
    task_chart_csv
):
    """Process uploaded document with selected tasks."""

    if file is None:
        return "Please upload a file first.", None, None, None

    if converter is None:
        return "Models not initialized. Please initialize first.", None, None, None

    # Determine which tasks to run
    selected_tasks = []
    if task_table_html:
        selected_tasks.append("Table (HTML)")
    if task_table_json:
        selected_tasks.append("Table (JSON)")
    if task_kvp:
        selected_tasks.append("Key-Value Pairs (KVP)")
    if task_chart_csv:
        selected_tasks.append("Chart (CSV)")

    if not export_markdown and not selected_tasks:
        return "Please select at least one output format.", None, None, None

    try:
        # Save uploaded file temporarily
        # Generate a unique temporary filename
        temp_path = Path(f"temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf")
        with open(temp_path, "wb") as f:
            # file is already bytes when type="binary"
            if isinstance(file, bytes):
                f.write(file)
            else:
                f.write(file.read())

        # Convert document
        status_text = "📄 Converting document with Docling...\n"
        conv_result = converter.convert(str(temp_path))

        # Export markdown if requested
        markdown_output = None
        markdown_file = None
        if export_markdown:
            status_text += "✅ Exporting to Markdown...\n"
            markdown_output = conv_result.document.export_to_markdown()
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            markdown_file = Path(OUTPUT_DIR) / f"doc_{timestamp}.md"
            with open(markdown_file, "w", encoding="utf-8") as f:
                f.write(markdown_output)
            status_text += f"✅ Markdown saved: {markdown_file.name}\n\n"

        # Process each page
        results_html = "<div style='font-family: sans-serif;'>"
        results_html += f"<h2>📊 Processing Results</h2>"
        results_html += f"<p><strong>Document:</strong> {temp_path.name}</p>"
        results_html += f"<p><strong>Total Pages:</strong> {len(conv_result.document.pages)}</p>"
        results_html += "<hr>"

        all_outputs = []

        for page_no, page_obj in conv_result.document.pages.items():
            status_text += f"\n📍 Processing Page {page_no}...\n"
            results_html += f"<h3>📄 Page {page_no}</h3>"

            if page_obj.image:
                page_img = page_obj.image.pil_image

                # Save page image for preview
                img_path = Path(OUTPUT_DIR) / f"page_{page_no}_{datetime.now().strftime('%H%M%S')}.png"
                page_img.save(img_path)

                results_html += f"<img src='file/{img_path}' style='max-width: 400px; border: 1px solid #ddd; margin: 10px 0;'><br>"

                # Run selected tasks
                for task_name in selected_tasks:
                    task_info = TASK_MAP[task_name]
                    status_text += f"  {task_info['icon']} Running {task_name}...\n"

                    result = run_vlm_task(page_img, task_info["prompt"])

                    # Save result
                    timestamp = datetime.now().strftime('%H%M%S')
                    out_name = f"pg{page_no}_{task_name.replace(' ', '_')}_{timestamp}.{task_info['ext']}"
                    out_path = Path(OUTPUT_DIR) / out_name

                    with open(out_path, "w", encoding="utf-8") as f:
                        f.write(result)

                    all_outputs.append(str(out_path))

                    # Add to results HTML
                    results_html += f"<div style='margin: 10px 0; padding: 10px; background: #f5f5f5; border-radius: 5px;'>"
                    results_html += f"<strong>{task_info['icon']} {task_name}</strong> "
                    results_html += f"<span style='color: #666; font-size: 0.9em;'>({out_name})</span><br>"

                    # Preview result
                    preview = result[:500] + "..." if len(result) > 500 else result
                    if task_info['ext'] == 'html':
                        results_html += f"<div style='margin-top: 5px; padding: 5px; background: white; border: 1px solid #ddd; max-height: 200px; overflow: auto;'>{preview}</div>"
                    else:
                        results_html += f"<pre style='margin-top: 5px; padding: 5px; background: white; border: 1px solid #ddd; max-height: 200px; overflow: auto; font-size: 0.85em;'>{preview}</pre>"

                    results_html += "</div>"

                    status_text += f"  ✅ {task_name} complete\n"

            results_html += "<hr>"

        results_html += "</div>"

        status_text += f"\n🎉 All tasks completed!\n"
        status_text += f"📁 Output directory: {OUTPUT_DIR}\n"
        status_text += f"📄 Total files generated: {len(all_outputs) + (1 if markdown_file else 0)}\n"

        # Clean up temp file
        if temp_path.exists():
            temp_path.unlink()

        return status_text, results_html, markdown_output, str(markdown_file) if markdown_file else None

    except Exception as e:
        error_msg = f"❌ Error processing document:\n{str(e)}\n\n{traceback.format_exc()}"
        return error_msg, None, None, None


def clear_output_folder():
    """Clear all files in the output folder."""
    try:
        count = 0
        for file in Path(OUTPUT_DIR).glob("*"):
            if file.is_file():
                file.unlink()
                count += 1
        return f"✅ Cleared {count} files from output folder"
    except Exception as e:
        return f"❌ Error clearing folder: {str(e)}"


def create_interface():
    """Create the Gradio interface."""

    with gr.Blocks(
        title="Granite 4.0 Vision + Docling Enhanced",
        theme=gr.themes.Soft(),
        css="""
        .output-box { border: 2px solid #e0e0e0; border-radius: 8px; padding: 15px; }
        .status-box { font-family: monospace; font-size: 0.9em; }
        """
    ) as app:

        gr.Markdown("""
        # 🚀 Granite 4.0 Vision + Docling Enhanced

        **Advanced document processing with IBM Granite 4.0 3B Vision and Docling**

        Upload a PDF or image, select your desired output formats, and let AI extract structured data!
        """)

        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### ⚙️ Configuration")

                init_btn = gr.Button("🔄 Initialize Models", variant="primary", size="lg")
                init_status = gr.Textbox(
                    label="Initialization Status",
                    lines=3,
                    interactive=False
                )

                gr.Markdown("---")
                gr.Markdown("### 📤 Upload Document")

                file_input = gr.File(
                    label="Upload PDF or Image",
                    file_types=[".pdf", ".png", ".jpg", ".jpeg"],
                    type="binary"
                )

                gr.Markdown("### 📋 Output Formats")

                export_markdown = gr.Checkbox(
                    label="📝 Export Full Markdown (Docling Native)",
                    value=True
                )

                gr.Markdown("**Granite AI Tasks:**")

                task_table_html = gr.Checkbox(
                    label="📊 Table (HTML)",
                    value=True
                )
                task_table_json = gr.Checkbox(
                    label="📋 Table (JSON)",
                    value=False
                )
                task_kvp = gr.Checkbox(
                    label="🔑 Key-Value Pairs (KVP)",
                    value=False
                )
                task_chart_csv = gr.Checkbox(
                    label="📈 Chart (CSV)",
                    value=False
                )

                gr.Markdown("---")

                process_btn = gr.Button("🚀 Start Processing", variant="primary", size="lg")
                clear_btn = gr.Button("🗑️ Clear Output Folder", variant="secondary")

                clear_status = gr.Textbox(
                    label="Clear Status",
                    lines=1,
                    interactive=False
                )

            with gr.Column(scale=2):
                gr.Markdown("### 📊 Processing Status")

                status_output = gr.Textbox(
                    label="Status Log",
                    lines=10,
                    interactive=False,
                    elem_classes=["status-box"]
                )

                gr.Markdown("### 🔍 Results Preview")

                with gr.Tabs():
                    with gr.Tab("📄 Visual Results"):
                        results_html = gr.HTML(
                            label="Results with Previews",
                            elem_classes=["output-box"]
                        )

                    with gr.Tab("📝 Markdown Output"):
                        markdown_output = gr.Textbox(
                            label="Full Markdown Content",
                            lines=20,
                            interactive=False
                        )
                        markdown_file = gr.Textbox(
                            label="Markdown File Path",
                            interactive=False
                        )

        gr.Markdown("""
        ---
        ### 📚 How to Use

        1. **Initialize Models**: Click "Initialize Models" (first time only, takes a few minutes)
        2. **Upload Document**: Choose a PDF or image file
        3. **Select Formats**: Check the output formats you want
        4. **Process**: Click "Start Processing" and wait for results
        5. **Review**: Check the results preview and status log
        6. **Access Files**: All outputs are saved in the `output/` folder

        ### 💡 Tips

        - **Markdown Export**: Best for full document conversion
        - **Table (HTML)**: Great for web display and styling
        - **Table (JSON)**: Perfect for data processing and APIs
        - **KVP**: Extracts key-value pairs from forms and documents
        - **Chart (CSV)**: Converts charts to structured data

        ### 🎯 Output Location

        All generated files are saved in: `output/` directory with timestamps
        """)

        # Event handlers
        init_btn.click(
            fn=initialize_models,
            outputs=[init_status]
        )

        process_btn.click(
            fn=process_document,
            inputs=[
                file_input,
                export_markdown,
                task_table_html,
                task_table_json,
                task_kvp,
                task_chart_csv
            ],
            outputs=[
                status_output,
                results_html,
                markdown_output,
                markdown_file
            ]
        )

        clear_btn.click(
            fn=clear_output_folder,
            outputs=[clear_status]
        )

    return app


def main():
    """Main entry point."""
    print(f"Starting Granite 4.0 Vision + Docling Enhanced UI")
    print(f"Device: {DEVICE}")
    print(f"Output directory: {OUTPUT_DIR}")

    app = create_interface()
    app.launch(
        server_name="0.0.0.0",
        server_port=7861,
        share=False,
        show_error=True
    )


if __name__ == "__main__":
    main()

# Made with Bob

Conclusion

The integration of Docling and Granite 4.0 3B Vision transforms unstructured documents into a strategic asset. By bridging the gap between raw visual data and structured digital information, this pipeline moves beyond simple text extraction to provide deep contextual understanding.

The samples provided in this post are intended to be used as productivity accelerators to help you build your own solutions!

>>> Thanks for reading <<<

Testing ‘chunkweaver’

Alain Airom (Ayrom) — Tue, 31 Mar 2026 11:33:38 +0000

Testing an open-source library I just read about based on the work of Oleksii Aleksapolskyi!

Introduction and Disclaimer

I’m always on the lookout for innovative open-source projects, and the article ‘Right Document, Wrong Chunk’ immediately caught my eye. After exploring the author’s repository and finding the approach incredibly compelling, I decided to build a custom interface and application to experiment with it firsthand.

Disclaimer: A huge shout-out to Oleksii Aleksapolskyi (Medium / LinkedIn), whose original research and backbone work made this project possible.

Excerpt from the author’s article;

Most RAG tutorials stop at: split, embed, retrieve top-K. Works fine on a README. Point the same pipeline at a 200-page regulation, a pile of clinical notes, or an RFC. Retrieval looks fine. Recall@5 is fine. MRR is fine. Users still complain.
Often it isn’t the wrong document. It’s the chunk: Article 17 cut in half, two RFC sections welded together, a table header orphaned from its rows. The model gets text that sort of matches the query and sort of doesn’t contain the answer. Right document. Wrong chunk. Standard metrics mostly won’t flag that.

Image from the author Oleksii Aleksapolskyi

The article gives the link to the GitHub repositry of “chunkweaver” and based of the samples provided on that repository, I built the application for my own tests!

The sample architecture schema provided by the author explains well the purpose of his implementation!

 PDF / DOCX / HTML              Your vector DB
        │                              ▲
        ▼                              │
  ┌───────────┐    ┌──────────────┐    │
  │ Extractor │───▶│ chunkweaver  │────┘
  │           │    │              │
  │ unstructured   │ boundaries   │  Embed + upsert
  │ marker-pdf     │ detectors    │  into Pinecone,
  │ docling        │ presets      │  Qdrant, Weaviate,
  │ pdfminer       │ overlap      │  ChromaDB, etc.
  └───────────┘    └──────────────┘

More interestingly, what I found really great on the GitHub repository are the following paragraphs;

Features

Zero dependencies — stdlib only, no LangChain/LlamaIndex tax
Regex boundaries — you tell the chunker where sections start (^Article \d+, ^## , ^Item 1.)
Hierarchical levels — CHAPTER > Section > Article > clause; split only as deep as needed
Heuristic detectors — HeadingDetector, TableDetector discover structure from text patterns
Annotation ingestion — accept pre-computed structure from any extractor
Semantic overlap — sentences, not characters
Full metadata — offsets, boundary types, hierarchy levels, overlap tracking
Integrations — LangChain and LlamaIndex drop-ins_

Presets

Built-in boundary patterns for common document types:

from chunkweaver.presets import (
    LEGAL_EU, LEGAL_US, RFC, MARKDOWN,
    CHAT, CLINICAL, FINANCIAL, FINANCIAL_TABLE,
    SEC_10K, FDA_LABEL, PLAIN,
)

| Preset            | Domain             | Detects                                           |
| ----------------- | ------------------ | ------------------------------------------------- |
| `LEGAL_EU`        | EU legislation     | `Article N`, `CHAPTER`, `SECTION`, `(1)` recitals |
| `LEGAL_US`        | US law / contracts | `§ N`, `Section N`, `WHEREAS`, `1.1` clauses      |
| `RFC`             | IETF RFCs          | `1. Intro`, `3.1 Overview`, `Appendix A`          |
| `MARKDOWN`        | Markdown           | `# headings`, `---` rules                         |
| `CHAT`            | Chat logs          | `[14:30]`, ISO timestamps, `speaker:` turns       |
| `CLINICAL`        | Medical notes      | `HPI:`, `ASSESSMENT:`, `PLAN:`, etc.              |
| `FINANCIAL`       | SEC filings        | `Item 1.`, `PART I`, `NOTE 1`, `Schedule A`       |
| `FINANCIAL_TABLE` | Data tables        | `TABLE N`, markdown/ASCII separators              |
| `SEC_10K`         | SEC annual reports | `PART I`–`IV`, `Item N.`, ALL-CAPS sub-headings   |
| `FDA_LABEL`       | Drug labels        | `1 INDICATIONS`, `## 2.1 Adult Dosage`            |
| `PLAIN`           | Any                | No boundaries — pure paragraph/sentence fallback  |

My local Implementation

My codebase in order to use and test the “chunkweaver” library, is based on;

Streamlit UI application,
Milvus Vector Database,
Ollama for local chat with the RAG’s content,
And sample data from public sources (At this point I didn’t implement any document transformation from various formats to text format yet, I used original text documents, but I saw the the author mentioned Docling as a solution among the document tranformation tools 🤗).

This gives the following architecture for my application ⬇️

The main application

As mentioned above, I built a Python application using Streamlit for the UI which proivdes document uploading and/or batch processing capacities. The chuncked data is then stored in my local Milvus instance.

# app.py
import os
import requests
import streamlit as st
from pathlib import Path

# Chunking & Embedding
from chunkweaver import Chunker
from chunkweaver.presets import LEGAL_EU
from chunkweaver.detector_heading import HeadingDetector
from chunkweaver.detector_table import TableDetector
from sentence_transformers import SentenceTransformer

# Vector DB
from pymilvus import MilvusClient, DataType, CollectionSchema, FieldSchema

# --- CONFIGURATION ---
INPUT_DIR = Path("input")
OUTPUT_DIR = Path("output")
INPUT_DIR.mkdir(exist_ok=True)
OUTPUT_DIR.mkdir(exist_ok=True)

COLLECTION_NAME = "chunkweaver_rag"
MODEL_NAME = 'all-MiniLM-L6-v2'
DIMENSION = 384 # Dimension for the MiniLM model

# --- RESOURCE INITIALIZATION ---
@st.cache_resource
def load_resources():
    # Load Embedding Model
    embedder = SentenceTransformer(MODEL_NAME)
    # Connect to Local Milvus
    client = MilvusClient(uri="http://localhost:19530")
    return embedder, client

embedder, milvus_client = load_resources()

# --- MILVUS ---
def init_milvus_collection():
    """Ensures collection exists, has an index, and is loaded."""
    if not milvus_client.has_collection(COLLECTION_NAME):
        from pymilvus import CollectionSchema, FieldSchema
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=DIMENSION),
            FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=500),
            FieldSchema(name="boundary_type", dtype=DataType.VARCHAR, max_length=100),
        ]
        schema = CollectionSchema(fields, description="Chunkweaver RAG collection")
        milvus_client.create_collection(collection_name=COLLECTION_NAME, schema=schema)

        # --- NEW: Create Index ---
        index_params = milvus_client.prepare_index_params()
        index_params.add_index(
            field_name="vector",
            index_type="HNSW",      
            metric_type="L2",       
            params={"M": 8, "efConstruction": 64}
        )
        milvus_client.create_index(COLLECTION_NAME, index_params)
        st.success("Collection and Index created!")


    milvus_client.load_collection(COLLECTION_NAME)

def run_indexing(file_paths, target_size, overlap):
    init_milvus_collection()

    chunker = Chunker(
        target_size=target_size,
        overlap=overlap,
        boundaries=LEGAL_EU,
        detectors=[HeadingDetector(), TableDetector()]
    )

    all_records = []
    for file_path in file_paths:
        text = file_path.read_text(encoding="utf-8")
        # Process using chunkweaver's metadata-rich chunking
        chunks = chunker.chunk_with_metadata(text)

        for i, c in enumerate(chunks):
            # Save raw chunk to output folder for inspection
            out_file = OUTPUT_DIR / f"{file_path.stem}_chunk_{i}.txt"
            out_file.write_text(c.text, encoding="utf-8")

            # Create embedding
            vector = embedder.encode(c.text).tolist()

            all_records.append({
                "vector": vector,
                "text": c.text,
                "source": file_path.name,
                "boundary_type": c.boundary_type
            })

    if all_records:
        res = milvus_client.insert(collection_name=COLLECTION_NAME, data=all_records)
        return len(all_records)
    return 0

# --- OLLAMA ---
def get_ollama_models():
    try:
        response = requests.get("http://localhost:11434/api/tags")
        return [m['name'] for m in response.json().get('models', [])]
    except:
        return []

def query_ollama(model, prompt):
    try:
        response = requests.post(
            "http://localhost:11434/api/generate",
            json={"model": model, "prompt": prompt, "stream": False},
            timeout=60
        )
        return response.json().get("response", "Error: No response field.")
    except Exception as e:
        return f"Error connecting to Ollama: {e}"

# --- STREAMLIT ---
st.set_page_config(page_title="Chunkweaver RAG", layout="wide")
st.title("🧶 Chunkweaver + Local Milvus + Ollama")

tab1, tab2, tab3 = st.tabs(["📁 Ingest Documents", "🔍 Milvus Data Viewer", "💬 RAG Chat"])

# TAB 1: INGESTION
with tab1:
    col1, col2 = st.columns([1, 2])
    with col1:
        st.subheader("Settings")
        t_size = st.number_input("Target Chunk Size (chars)", 128, 4096, 1024)
        ov_size = st.slider("Sentence Overlap", 0, 5, 2)

        st.divider()
        uploaded_files = st.file_uploader("Upload .txt files", type=['txt'], accept_multiple_files=True)
        mode = st.radio("Processing Mode", ["Only Process Newly Uploaded", "Batch Process ALL in 'input/' folder"])

        if st.button("🚀 Run Chunkweaver & Index"):
            files_to_process = []
            if mode == "Only Process Newly Uploaded":
                if not uploaded_files:
                    st.error("No files uploaded.")
                for f in uploaded_files:
                    p = INPUT_DIR / f.name
                    p.write_bytes(f.getbuffer())
                    files_to_process.append(p)
            else:
                files_to_process = list(INPUT_DIR.glob("*.txt"))

            if files_to_process:
                with st.spinner(f"Processing {len(files_to_process)} files..."):
                    count = run_indexing(files_to_process, t_size, ov_size)
                    st.success(f"Success! {count} chunks added to Milvus.")
            else:
                st.warning("No files found to process.")

    with col2:
        st.subheader("Files in 'input/' folder")
        existing_files = list(INPUT_DIR.glob("*.txt"))
        if existing_files:
            for f in existing_files:
                st.text(f"📄 {f.name} ({f.stat().st_size / 1024:.1f} KB)")
        else:
            st.info("The input folder is currently empty.")

# TAB 2: VIEWER
with tab2:
    st.subheader("Current Data in Milvus")
    if st.button("Refresh View"):
        try:
            if milvus_client.has_collection(COLLECTION_NAME):
                milvus_client.load_collection(COLLECTION_NAME)
                results = milvus_client.query(
                    collection_name=COLLECTION_NAME,
                    filter="",
                    limit=20,
                    output_fields=["source", "boundary_type", "text"]
                )
                if results:
                    st.table(results)
                else:
                    st.info("Collection is empty.")
            else:
                st.warning("Collection not created yet. Please index some documents.")
        except Exception as e:
            st.error(f"Milvus Error: {e}")

# TAB 3: CHAT
with tab3:
    st.subheader("Chat with your Data")
    models = get_ollama_models()
    if not models:
        st.error("Ollama not detected. Please ensure 'ollama serve' is running.")
    else:
        selected_model = st.selectbox("Select your Local Model", models)
        user_input = st.text_input("Ask a question about your documents:")

        if user_input:
            try:
                # 1. Retrieval
                milvus_client.load_collection(COLLECTION_NAME)
                query_vec = embedder.encode(user_input).tolist()
                search_results = milvus_client.search(
                    collection_name=COLLECTION_NAME,
                    data=[query_vec],
                    limit=3,
                    output_fields=["text", "source"]
                )

                # 2. Context Construction
                retrieved_chunks = []
                for hits in search_results:
                    for hit in hits:
                        retrieved_chunks.append(hit['entity']['text'])

                context = "\n---\n".join(retrieved_chunks)

                # 3. Generation
                full_prompt = f"""Use the following context to answer the user's question. 
                If the answer isn't in the context, say you don't know.

                Context:
                {context}

                Question: {user_input}

                Answer:"""

                with st.spinner("Ollama generating response..."):
                    answer = query_ollama(selected_model, full_prompt)
                    st.markdown(f"### Answer\n{answer}")

                with st.expander("View Source Context"):
                    st.write(context)

            except Exception as e:
                st.error(f"RAG Error: {e}")

Install the requirements.txt

pip install -r requirements.txt

chunkweaver
streamlit
pymilvus 
sentence-transformers

Milvus Vector DB Implementation

To use Milvus locally, I use Podman alongside with Podman Desktop.

Create the local Milvus folder(s)

mkdir volumes && mkdir volumes/milvus

Download Milvus Images and Run them

# Download the official standalone script
curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh

# Replace 'docker' with 'podman' for compatibility
sed -i 's/docker/podman/g' standalone_embed.sh
# or if you have an error message on a MacOS;
sed -i '' 's/docker/podman/g' standalone_embed.sh

# Start Milvus
bash standalone_embed.sh start

Once the code has ran, you’ll get the following structure (or something like it!) ⬇️

As I tried several times creating chunks I used the following script to test my Milvus connection and remove the collections I already made (very basic but works fine).

from pymilvus import MilvusClient

try:
    MilvusClient.connect(host="127.0.0.1", port="19530")
    print("Milvus connection successful!")
    connections.disconnect("default")
except Exception as e:
    print(f"Milvus connection failed: {e}")


client = MilvusClient(uri="http://localhost:19530")
if client.has_collection("chunkweaver_rag"):
    client.drop_collection("chunkweaver_rag")
    print("Dropped broken collection.")

You’ll also have the chunks outputs 📩

2024/1689

12.7.2024

REGULATION (EU) 2024/1689 OF THE EUROPEAN PARLIAMENT AND OF THE COUNCIL

of 13 June 2024

laying down harmonised rules on artificial intelligence and amending Regulations (EC) No 300/2008, (EU) No 167/2013, (EU) No 168/2013, (EU) 2018/858, (EU) 2018/1139 and (EU) 2019/2144 and Directives 2014/90/EU, (EU) 2016/797 and (EU) 2020/1828 (Artificial Intelligence Act)

(Text with EEA relevance)

The chat implementation using Ollama

As always, I rely on my local Ollama setup alongside with my downloaded models. The UI let’s the user pick any installed LLM.

Documents I used to enrich the RAG

Trying to be as close as possible to the original thaoughts of the implementation, I used the following sites/documents and created two text format files in order to chuck and embed in Milvus vector database.

Eur-lex — Access to European Union law (Regulation (EU) 2024/1689 of the European Parliament and of the Council of 13 June 2024 laying down harmonised rules on artificial intelligence and amending Regulations (EC) No 300/2008, (EU) No 167/2013, (EU) No 168/2013, (EU) 2018/858, (EU) 2018/1139 and (EU) 2019/2144 and Directives 2014/90/EU, (EU) 2016/797 and (EU) 2020/1828 (Artificial Intelligence Act) (Text with EEA relevance)): https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:32024R1689
United Nations site (Universal Declaration of Human Rights): https://www.un.org/en/about-us/universal-declaration-of-human-rights

Conclusion

In conclusion, exploring and implementing the chunkweaver library has been an eye-opening experience. The results demonstrate that moving beyond rigid, fixed-size splitting toward a structure-aware approach is not just a marginal improvement, but a fundamental shift in ensuring RAG reliability. This project has truly given me a new perspective on how we should handle document ingestion — prioritizing the ‘natural’ boundaries of information to preserve the author’s original context. Huge kudos to Oleksii Aleksapolskyi for his innovative work and for providing the community with such a promising tool; it has significantly changed how I approach building high-quality retrieval systems.

>>> Thanks for reading <<<

Building Smarter Agents: A Guide to IBM Bob’s “MCP Builder with Agent Utils”

Alain Airom (Ayrom) — Mon, 30 Mar 2026 10:11:06 +0000

Using a new “Mode” in IBM Bob to create MCP/Agent utils!

Building Smarter Agents: A Guide to IBM’s “MCP Builder with Agent Utils”

In the evolving landscape of AI, the ability for agents to interact with the real world is a game-changer. IBM’s agent’s util library implemented in Bob IDE, often referred to in the context of the "MCP Builder," provides a robust framework for creating Model Context Protocol (MCP) servers. These servers act as a bridge, allowing AI agents (like Claude or GPT-4 or others… you want it, yon name it 😀) to discover and use external tools to perform complex tasks, such as retrieving real-time weather data or performing unit conversions.

The Foundation: BaseMCP and the Tool Registry

At the heart of any MCP server built with these utilities is the BaseMCP class. This base class handles the heavy lifting of the protocol, including tool registration, request routing, and life-cycle management. By extending BaseMCP, developers can focus entirely on the logic of their tools rather than the underlying communication infrastructure.

To expose a Python method as a tool that an agent can call, you simply use the @tool decorator. This decorator automatically registers the method in the server's tool registry, making it discoverable by any connected MCP client.

Anatomy of an MCP Tool

Creating a tool is straightforward but follows a specific pattern to ensure compatibility and performance. Every tool must be an async method and include a Context object as its first parameter. This Context object provides request-scoped information, such as request IDs and metadata, which is essential for logging and tracing in production environments.

A typical tool implementation looks like this:

Decorator: @tool marks the function for the AI.
Signature: async def get_current_weather(self, ctx: Context, location: str, ...).
Docstring: A comprehensive description that tells the AI when and how to use the tool.
Return: A dictionary containing a success flag and the resulting data or an error message.

"""
Weather MCP Server - A sample implementation using aicoe-agent-utils

This MCP server provides weather-related tools that can be used by AI agents.
It demonstrates the proper structure and patterns for building MCP servers
with the aicoe-agent-utils library.
"""

import os
import asyncio
from typing import Dict, Any, Optional
from datetime import datetime
import httpx
from dotenv import load_dotenv

# Import from aicoe-agent-utils
# For demonstration without access to the actual library, we use a mock implementation
try:
    from aicoe_agent_utils.mcp.base_mcp import BaseMCP, tool, Context
except ImportError:
    # Fall back to mock implementation
    from mock_agent_utils import BaseMCP, tool, Context

# Load environment variables
load_dotenv()


class WeatherMCPServer(BaseMCP):
    """
    Weather MCP Server providing weather information tools.

    This server extends BaseMCP and provides tools for:
    - Getting current weather for a location
    - Getting weather forecast
    - Converting temperature units
    - Getting weather alerts
    """

    def __init__(self):
        """Initialize the Weather MCP Server."""
        super().__init__()
        self.api_key = os.getenv("WEATHER_API_KEY", "demo_key")
        self.base_url = "https://api.openweathermap.org/data/2.5"

    @tool
    async def get_current_weather(
        self,
        ctx: Context,
        location: str,
        units: str = "metric"
    ) -> Dict[str, Any]:
        """
        Get current weather information for a specified location.

        Args:
            ctx: The context object provided by the MCP framework
            location: City name or coordinates (e.g., "London" or "51.5074,-0.1278")
            units: Temperature units - "metric" (Celsius), "imperial" (Fahrenheit), or "kelvin"

        Returns:
            Dictionary containing:
                - temperature: Current temperature
                - feels_like: Feels like temperature
                - humidity: Humidity percentage
                - description: Weather description
                - wind_speed: Wind speed
                - timestamp: Time of data calculation
        """
        try:
            # In a real implementation, this would call the weather API
            # For demo purposes, we'll return mock data
            async with httpx.AsyncClient() as client:
                # Mock response for demonstration
                weather_data = {
                    "location": location,
                    "temperature": 22.5 if units == "metric" else 72.5,
                    "feels_like": 21.0 if units == "metric" else 69.8,
                    "humidity": 65,
                    "description": "Partly cloudy",
                    "wind_speed": 5.2 if units == "metric" else 11.6,
                    "units": units,
                    "timestamp": datetime.now().isoformat()
                }

                return {
                    "success": True,
                    "data": weather_data
                }

        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    @tool
    async def get_weather_forecast(
        self,
        ctx: Context,
        location: str,
        days: int = 5,
        units: str = "metric"
    ) -> Dict[str, Any]:
        """
        Get weather forecast for a specified location.

        Args:
            ctx: The context object provided by the MCP framework
            location: City name or coordinates
            days: Number of days to forecast (1-7)
            units: Temperature units - "metric", "imperial", or "kelvin"

        Returns:
            Dictionary containing:
                - location: Location name
                - forecast: List of daily forecasts with temperature, conditions, etc.
                - units: Temperature units used
        """
        try:
            if days < 1 or days > 7:
                return {
                    "success": False,
                    "error": "Days must be between 1 and 7"
                }

            # Mock forecast data
            forecast_data = []
            for i in range(days):
                forecast_data.append({
                    "day": i + 1,
                    "date": datetime.now().isoformat(),
                    "temp_high": 25 + i if units == "metric" else 77 + i * 1.8,
                    "temp_low": 15 + i if units == "metric" else 59 + i * 1.8,
                    "description": "Sunny" if i % 2 == 0 else "Cloudy",
                    "precipitation_chance": 20 + i * 10
                })

            return {
                "success": True,
                "data": {
                    "location": location,
                    "forecast": forecast_data,
                    "units": units
                }
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    @tool
    async def convert_temperature(
        self,
        ctx: Context,
        temperature: float,
        from_unit: str,
        to_unit: str
    ) -> Dict[str, Any]:
        """
        Convert temperature between different units.

        Args:
            ctx: The context object provided by the MCP framework
            temperature: Temperature value to convert
            from_unit: Source unit - "celsius", "fahrenheit", or "kelvin"
            to_unit: Target unit - "celsius", "fahrenheit", or "kelvin"

        Returns:
            Dictionary containing:
                - original_value: Input temperature
                - original_unit: Input unit
                - converted_value: Converted temperature
                - converted_unit: Output unit
        """
        try:
            # Conversion logic
            from_unit = from_unit.lower()
            to_unit = to_unit.lower()

            # Convert to Celsius first
            if from_unit == "celsius":
                celsius = temperature
            elif from_unit == "fahrenheit":
                celsius = (temperature - 32) * 5/9
            elif from_unit == "kelvin":
                celsius = temperature - 273.15
            else:
                return {
                    "success": False,
                    "error": f"Invalid from_unit: {from_unit}"
                }

            # Convert from Celsius to target unit
            if to_unit == "celsius":
                result = celsius
            elif to_unit == "fahrenheit":
                result = (celsius * 9/5) + 32
            elif to_unit == "kelvin":
                result = celsius + 273.15
            else:
                return {
                    "success": False,
                    "error": f"Invalid to_unit: {to_unit}"
                }

            return {
                "success": True,
                "data": {
                    "original_value": temperature,
                    "original_unit": from_unit,
                    "converted_value": round(result, 2),
                    "converted_unit": to_unit
                }
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    @tool
    async def get_weather_alerts(
        self,
        ctx: Context,
        location: str
    ) -> Dict[str, Any]:
        """
        Get weather alerts and warnings for a specified location.

        Args:
            ctx: The context object provided by the MCP framework
            location: City name or coordinates

        Returns:
            Dictionary containing:
                - location: Location name
                - alerts: List of active weather alerts
                - alert_count: Number of active alerts
        """
        try:
            # Mock alert data
            alerts = [
                {
                    "type": "Heat Advisory",
                    "severity": "Moderate",
                    "description": "High temperatures expected",
                    "start_time": datetime.now().isoformat(),
                    "end_time": datetime.now().isoformat()
                }
            ]

            return {
                "success": True,
                "data": {
                    "location": location,
                    "alerts": alerts,
                    "alert_count": len(alerts)
                }
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    async def test(self):
        """
        Test method to verify all tools work correctly.
        This method is optional but recommended for local testing.
        """
        print("Testing Weather MCP Server...")
        print("-" * 50)

        # Create a mock context
        ctx = Context()

        # Test get_current_weather
        print("\n1. Testing get_current_weather...")
        result = await self.get_current_weather(ctx, "London", "metric")
        print(f"Result: {result}")

        # Test get_weather_forecast
        print("\n2. Testing get_weather_forecast...")
        result = await self.get_weather_forecast(ctx, "Paris", 3, "metric")
        print(f"Result: {result}")

        # Test convert_temperature
        print("\n3. Testing convert_temperature...")
        result = await self.convert_temperature(ctx, 25, "celsius", "fahrenheit")
        print(f"Result: {result}")

        # Test get_weather_alerts
        print("\n4. Testing get_weather_alerts...")
        result = await self.get_weather_alerts(ctx, "New York")
        print(f"Result: {result}")

        print("\n" + "-" * 50)
        print("All tests completed!")


if __name__ == "__main__":
    import sys

    # Create server instance
    mcp = WeatherMCPServer()

    # Check if running in test mode
    if len(sys.argv) > 1 and sys.argv[1] == "test":
        # Run tests
        asyncio.run(mcp.test())
    else:
        # Start the MCP server
        print("Starting Weather MCP Server...")
        print(f"Server will be available at: http://localhost:{os.getenv('WEATHER_MCP_PORT', 8000)}/mcp")
        mcp.run()

# Made with Bob

From Development to Production

One of the standout features of this toolkit is the ease of testing. The framework supports a “test mode” where developers can run their tools locally using a built-in test() method before ever deploying to a live environment. This ensures that the logic is sound and the AI-facing documentation is accurate.

For those without immediate access to internal IBM repositories, the project even includes a mock_agent_utils.py implementation. This mock mimics the real library's interface using FastAPI, allowing developers to build and verify their MCP servers entirely offline or in restricted environments.

"""
Mock implementation of aicoe-agent-utils for demonstration purposes.

This file provides mock implementations of the BaseMCP framework components
to allow the Weather MCP Server to run without requiring access to the actual
aicoe-agent-utils library.

In a real implementation, you would use the actual aicoe-agent-utils library
from: https://github.ibm.com/AI-CoE/aicoe-agent-utils
"""

import asyncio
import inspect
import logging
import os
from datetime import datetime
from functools import wraps
from typing import Any, Callable, Dict, Optional
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import uvicorn

# Configure logging
logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO"),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class Context:
    """
    Mock Context object that provides request-scoped information to tools.

    In the real aicoe-agent-utils, this would include additional features
    like user authentication, session management, and distributed tracing.
    """

    def __init__(self):
        self.request_id: str = f"req_{datetime.now().timestamp()}"
        self.timestamp: datetime = datetime.now()
        self.user_id: Optional[str] = None
        self.session_id: Optional[str] = None
        self.metadata: Dict[str, Any] = {}


def tool(func: Callable) -> Callable:
    """
    Mock @tool decorator that marks methods as MCP tools.

    In the real aicoe-agent-utils, this decorator would:
    - Extract parameter metadata
    - Register the tool in the MCP registry
    - Add validation and error handling
    - Generate OpenAPI documentation
    """

    @wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)

    # Mark the function as a tool
    wrapper._is_mcp_tool = True
    wrapper._tool_name = func.__name__
    wrapper._tool_doc = func.__doc__ or ""

    return wrapper


class BaseMCP:
    """
    Mock BaseMCP base class for building MCP servers.

    In the real aicoe-agent-utils, this class would provide:
    - Automatic tool discovery and registration
    - Request routing and validation
    - Response formatting
    - Error handling and logging
    - Configuration management
    - Integration with FastMCP protocol
    """

    def __init__(self):
        """Initialize the MCP server."""
        self.app = FastAPI(title="MCP Server")
        self.tools: Dict[str, Callable] = {}
        self.port = int(os.getenv("WEATHER_MCP_PORT", "8000"))
        self.host = os.getenv("HOST", "0.0.0.0")

        # Discover and register tools
        self._discover_tools()
        self._setup_routes()

        logger.info("MCP Server initialized")

    def _discover_tools(self):
        """
        Discover all methods decorated with @tool.

        This scans the class for methods with the _is_mcp_tool attribute
        and registers them in the tools dictionary.
        """
        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            if hasattr(method, '_is_mcp_tool'):
                self.tools[method._tool_name] = method
                logger.info(f"Registered tool: {method._tool_name}")

    def _setup_routes(self):
        """Set up FastAPI routes for the MCP server."""

        @self.app.get("/mcp")
        async def health_check():
            """Health check endpoint."""
            return {
                "status": "ok",
                "server": "Weather MCP Server",
                "tools": list(self.tools.keys())
            }

        @self.app.get("/mcp/tools")
        async def list_tools():
            """List all available tools."""
            tools_info = []
            for name, method in self.tools.items():
                sig = inspect.signature(method)
                params = {}
                for param_name, param in sig.parameters.items():
                    if param_name not in ['self', 'ctx']:
                        params[param_name] = {
                            "type": str(param.annotation) if param.annotation != inspect.Parameter.empty else "any",
                            "required": param.default == inspect.Parameter.empty,
                            "default": None if param.default == inspect.Parameter.empty else param.default
                        }

                tools_info.append({
                    "name": name,
                    "description": method._tool_doc.split('\n')[0].strip(),
                    "parameters": params
                })

            return {"tools": tools_info}

        @self.app.post("/mcp/tools/{tool_name}")
        async def execute_tool(tool_name: str, parameters: Dict[str, Any]):
            """Execute a specific tool."""
            if tool_name not in self.tools:
                raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")

            try:
                # Create context
                ctx = Context()

                # Execute tool
                logger.info(f"Executing tool: {tool_name}", extra={"request_id": ctx.request_id})
                result = await self.tools[tool_name](ctx, **parameters)

                logger.info(f"Tool executed successfully: {tool_name}", extra={"request_id": ctx.request_id})
                return result

            except TypeError as e:
                logger.error(f"Invalid parameters for tool {tool_name}: {e}")
                raise HTTPException(status_code=422, detail=f"Invalid parameters: {str(e)}")
            except Exception as e:
                logger.error(f"Error executing tool {tool_name}: {e}", exc_info=True)
                raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

    def run(self):
        """
        Start the MCP server.

        In the real aicoe-agent-utils, this would integrate with the
        FastMCP protocol and provide additional features like:
        - WebSocket support
        - Server-Sent Events (SSE)
        - Authentication
        - Rate limiting
        """
        logger.info(f"Starting MCP Server on {self.host}:{self.port}")
        logger.info(f"Server will be available at: http://{self.host}:{self.port}/mcp")
        logger.info(f"Registered tools: {', '.join(self.tools.keys())}")

        uvicorn.run(
            self.app,
            host=self.host,
            port=self.port,
            log_level=os.getenv("LOG_LEVEL", "info").lower()
        )


# Export the mock implementations
__all__ = ['BaseMCP', 'tool', 'Context']

# Made with Bob

Best Practices for Scalable Agents

To build production-ready agents, the documentation emphasizes several best practices:

Non-blocking I/O: Always use async/await and asynchronous clients like httpx to prevent the server from hanging during API calls.
Type Hinting: Use Python type hints for all parameters so the MCP framework can validate inputs automatically.
Structured Logging: Leverage the built-in logger to track tool execution and debug issues in real-time.
Error Handling: Wrap tool logic in try-except blocks to return clean, standardized error responses to the agent.

python3 weather_mcp_server.py test
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: convert_temperature
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_current_weather
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_weather_alerts
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_weather_forecast
2026-03-30 10:42:27,882 - mock_agent_utils - INFO - MCP Server initialized
Testing Weather MCP Server...
--------------------------------------------------

1. Testing get_current_weather...
Result: {'success': True, 'data': {'location': 'London', 'temperature': 22.5, 'feels_like': 21.0, 'humidity': 65, 'description': 'Partly cloudy', 'wind_speed': 5.2, 'units': 'metric', 'timestamp': '2026-03-30T10:42:27.948120'}}

2. Testing get_weather_forecast...
Result: {'success': True, 'data': {'location': 'Paris', 'forecast': [{'day': 1, 'date': '2026-03-30T10:42:27.951672', 'temp_high': 25, 'temp_low': 15, 'description': 'Sunny', 'precipitation_chance': 20}, {'day': 2, 'date': '2026-03-30T10:42:27.951678', 'temp_high': 26, 'temp_low': 16, 'description': 'Cloudy', 'precipitation_chance': 30}, {'day': 3, 'date': '2026-03-30T10:42:27.951681', 'temp_high': 27, 'temp_low': 17, 'description': 'Sunny', 'precipitation_chance': 40}], 'units': 'metric'}}

3. Testing convert_temperature...
Result: {'success': True, 'data': {'original_value': 25, 'original_unit': 'celsius', 'converted_value': 77.0, 'converted_unit': 'fahrenheit'}}

4. Testing get_weather_alerts...
Result: {'success': True, 'data': {'location': 'New York', 'alerts': [{'type': 'Heat Advisory', 'severity': 'Moderate', 'description': 'High temperatures expected', 'start_time': '2026-03-30T10:42:27.951711', 'end_time': '2026-03-30T10:42:27.951712'}], 'alert_count': 1}}

--------------------------------------------------
All tests completed!

Conclusion

By following these patterns and using Bob (and specific modes), developers can rapidly move from a simple concept to a sophisticated, tool-augmented AI agent capable of navigating complex real-world workflows.

>>> Thanks for reading <<<

Discovering “GraphQLite” from Colliery-IO 📈

Alain Airom (Ayrom) — Thu, 26 Mar 2026 16:04:16 +0000

A rapid implementation/test of “GraphQLite” using IBM Bob (of course 😅)

While exploring new tools for efficient data architecture, I came across GraphQLite. I was immediately impressed by how it simplifies graph database implementation. With its well-documented GitHub repository and high-quality samples, it’s a ‘plug-and-play’ solution that allows you to start building right away.

Intrduction and Excerpt from GraphQLite Repository

GraphQLite is An SQLite extension that adds graph database capabilities using the Cypher query language.

Store and query graph data directly in SQLite — combining the simplicity of a single-file, zero-config embedded database with Cypher’s expressive power for modeling relationships, presenting the following features;

Cypher queries — MATCH, CREATE, MERGE, SET, DELETE, WITH, UNWIND, RETURN
Graph algorithms — PageRank, Louvain, Dijkstra, BFS/DFS, connected components, and more
Zero configuration — Works with any SQLite database, no server required
Multiple bindings — Python, Rust, and raw SQL interfaces.

Moving past the provided samples (working great by the way), I decided to put GraphQLite to the test. I got Bob busy building a simple UI to help me really dive into the implementation and see what it can do!

Moving Beyond Simple Search: Building with GraphQLite

Image from the owners of GitHub repository

As I wanted to see GraphQLite in action, I put Bob to work building a custom GraphRAG Knowledge System. It’s a system that takes raw documents and maps them into a living network of interconnected ideas. Instead of just “storing” a file, The implementation transforms text into a visual web where “Entities” which are linked by their “Relationships.” It turns a flat pile of data into a structured map that shows exactly how different concepts mention and influence one another.

To give this system a “brain,” Ollama handles the local natural language processing. This is where the magic happens: when you upload a .txt file, the system uses an LLM to automatically extract key entities and store them in a local SQLite-backed graph. When you head over to the "Ask Questions" panel, the system doesn't just look for keywords. It performs a hybrid search, combining traditional vector similarity with deep graph traversals. This ensures that when you get an answer, it’s not just a guess—it’s a grounded response backed by the specific context of your own data.

The best part of what Bob built is the transparency. Usually, RAG (Retrieval-Augmented Generation) feels like a “black box,” but this interface pulls back the curtain. Through the Knowledge Graph Visualization, you can actually watch the system work, monitoring real-time stats like the number of identified communities or document relations. Bob even implemented advanced algorithms like Louvain community detection to help group related data points automatically. It’s a powerful, “plug-and-play” way to turn a simple SQLite setup into a sophisticated knowledge engine that you can actually see and touch.

The Application

The application is based on two parts; a python code which does the processing, and a HTML file for the UI.

Processing and Back-End Engine

The app.py file serves as the "brain" of the operation, acting as a bridge between the GraphQLite database and the user interface. Its primary job is to manage the lifecycle of your data.

Initialization: It kicks things off by setting up the GraphQLite instance and connecting it to a local SQLite database (graph.db). It also initializes the Ollama integration, ensuring the system is ready to process natural language.
Data Ingestion: When you upload a file or paste text, the backend doesn’t just save it; it parses the content, identifies key entities, and maps out the “MENTIONS” relationships that turn a flat document into a connected graph.
The Hybrid Search: This is the most impressive part. When a question is asked, the script triggers a search that combines Vector Embeddings (for semantic similarity) with Graph Traversal (for structural context). This ensures the AI isn’t just guessing — it’s navigating a logical map of your data to find the right answer.

# Workaround for macOS SQLite extension loading issue
# Use pysqlite3 instead of built-in sqlite3
import sys
try:
    __import__('pysqlite3')
    sys.modules['sqlite3'] = sys.modules.pop('pysqlite3')
except ImportError:
    pass  # Fall back to built-in sqlite3

"""
GraphRAG Application with Web UI

A knowledge graph-based RAG system using GraphQLite, sqlite-vec, and Ollama.
Provides a web interface for document ingestion and question answering.
"""

import os
import struct
import logging
from pathlib import Path
from flask import Flask, render_template, request, jsonify, send_from_directory, g as flask_g

import numpy as np
from sentence_transformers import SentenceTransformer

from graphqlite import graph
from ollama_client import OllamaClient, Message

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Suppress library logging
os.environ["TOKENIZERS_PARALLELISM"] = "true"
logging.getLogger("sentence_transformers").setLevel(logging.WARNING)
logging.getLogger("transformers").setLevel(logging.WARNING)

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max file size

# Global configuration
DB_PATH = "knowledge_graph.db"
OLLAMA_MODEL = "ibm/granite4:3b"

# Thread-safe shared resources
embed_model = None
ollama_client = None

SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Instructions:
- Answer the question using ONLY the information in the context below
- Be concise and direct
- If the context doesn't contain enough information, say so
- For yes/no questions, start with "Yes" or "No" then explain briefly"""


def make_safe_id(s: str) -> str:
    """Create a safe node ID from a string."""
    return s.replace("'", "_").replace('"', "_").replace("\\", "_").replace("\n", " ")


def load_vec_extension(g):
    """Load sqlite-vec extension for vector similarity search."""
    import sqlite_vec

    sqlite_conn = g.connection.sqlite_connection
    sqlite_conn.enable_load_extension(True)
    sqlite_vec.load(sqlite_conn)
    sqlite_conn.enable_load_extension(False)

    # Create embedding table
    sqlite_conn.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS document_embeddings USING vec0(
            doc_id TEXT PRIMARY KEY,
            embedding FLOAT[384]
        )
    """)
    g.connection.commit()


def serialize_embedding(embedding: np.ndarray) -> bytes:
    """Serialize embedding to bytes for sqlite-vec."""
    return struct.pack(f"{len(embedding)}f", *embedding.astype(np.float32))


def deserialize_embedding(blob: bytes) -> np.ndarray:
    """Deserialize embedding from sqlite-vec bytes."""
    n = len(blob) // 4
    return np.array(struct.unpack(f"{n}f", blob), dtype=np.float32)


def get_graph():
    """Get or create a graph connection for the current request."""
    if 'graph_connection' not in flask_g:
        flask_g.graph_connection = graph(DB_PATH)
        load_vec_extension(flask_g.graph_connection)
    return flask_g.graph_connection


@app.teardown_appcontext
def close_graph_connection(error):
    """Close the graph connection at the end of each request."""
    graph_conn = flask_g.pop('graph_connection', None)
    if graph_conn is not None:
        graph_conn.close()


def get_communities() -> dict[str, int]:
    """Get or compute community detection results for current request."""
    if 'communities' not in flask_g:
        logger.info("Running community detection (Louvain)...")
        g = get_graph()
        results = g.louvain(resolution=1.0)
        communities = {}
        for r in results:
            node_id = r.get("user_id") or r.get("node_id")
            if node_id:
                communities[node_id] = r.get("community", -1)
        logger.info(f"Found {len(set(communities.values()))} communities")
        flask_g.communities = communities
    return flask_g.communities


def ingest_document(title: str, content: str) -> dict:
    """
    Ingest a document into the knowledge graph.

    Args:
        title: Document title
        content: Document content

    Returns:
        Dict with ingestion results
    """
    try:
        g = get_graph()

        # Create safe IDs
        doc_id = f"doc:{make_safe_id(title)}"
        safe_title = make_safe_id(title)
        safe_content = make_safe_id(content[:1000])  # Limit content length

        # Generate embedding
        embedding = embed_model.encode([content], convert_to_numpy=True)[0]
        embedding_bytes = serialize_embedding(embedding)

        # Create Document node
        g.upsert_node(
            doc_id,
            {"title": safe_title, "content": safe_content},
            "Document"
        )

        # Store embedding
        sqlite_conn = g.connection.sqlite_connection
        sqlite_conn.execute(
            "INSERT OR REPLACE INTO document_embeddings (doc_id, embedding) VALUES (?, ?)",
            (doc_id, embedding_bytes)
        )

        g.connection.commit()

        # Extract entities (simple keyword extraction)
        entities = extract_entities(content)
        entity_count = 0

        for entity in entities[:10]:  # Limit to 10 entities
            entity_id = f"entity:{make_safe_id(entity)}"
            g.upsert_node(
                entity_id,
                {"name": make_safe_id(entity)},
                "Entity"
            )

            # Link document to entity
            g.upsert_edge(
                doc_id,
                entity_id,
                {},
                "MENTIONS"
            )
            entity_count += 1

        g.connection.commit()

        # Clear community cache for this request
        flask_g.pop('communities', None)

        return {
            "success": True,
            "doc_id": doc_id,
            "title": title,
            "entities_extracted": entity_count
        }

    except Exception as e:
        logger.error(f"Error ingesting document: {e}")
        return {
            "success": False,
            "error": str(e)
        }


def extract_entities(text: str) -> list[str]:
    """Simple entity extraction (capitalized words)."""
    words = text.split()
    entities = []
    for word in words:
        # Simple heuristic: capitalized words that aren't at sentence start
        if word and word[0].isupper() and len(word) > 3:
            clean_word = word.strip('.,!?;:"()[]{}')
            if clean_word and clean_word not in entities:
                entities.append(clean_word)
    return entities[:20]  # Limit to 20 entities


def vector_search(query: str, k: int = 5) -> list[dict]:
    """Find top-k similar documents using vector search."""
    g = get_graph()
    query_embedding = embed_model.encode([query], convert_to_numpy=True)[0]
    query_bytes = serialize_embedding(query_embedding)

    sqlite_conn = g.connection.sqlite_connection
    cursor = sqlite_conn.execute("""
        SELECT doc_id, distance
        FROM document_embeddings
        WHERE embedding MATCH ?
        ORDER BY distance
        LIMIT ?
    """, (query_bytes, k))

    results = []
    for row in cursor:
        doc_id = row[0]
        distance = row[1]
        node = g.get_node(doc_id)
        if node:
            results.append({
                "id": doc_id,
                "distance": float(distance),
                "properties": node.get("properties", {})
            })
    return results


def get_related_documents(doc_id: str) -> list[dict]:
    """Get documents related via shared entities."""
    g = get_graph()
    query = f"""
    MATCH (d {{id: '{doc_id}'}})-[:MENTIONS]->(e:Entity)<-[:MENTIONS]-(related:Document)
    WHERE related.id <> '{doc_id}'
    RETURN DISTINCT related.id AS id, related.title AS title, related.content AS content
    LIMIT 5
    """

    results = []
    try:
        result = g.connection.cypher(query)
        for row in result:
            results.append({
                "id": row.get("id"),
                "title": row.get("title"),
                "content": row.get("content")
            })
    except Exception as e:
        logger.error(f"Error in graph traversal: {e}")

    return results


def get_community_documents(doc_id: str, limit: int = 3) -> tuple[list[dict], int]:
    """Get other documents in the same community."""
    g = get_graph()
    communities = get_communities()
    my_community = communities.get(doc_id, -1)

    if my_community == -1:
        return [], -1

    same_community = []
    for node_id, comm in communities.items():
        if comm == my_community and node_id != doc_id and node_id.startswith("doc:"):
            same_community.append(node_id)

    results = []
    for node_id in same_community[:limit]:
        node = g.get_node(node_id)
        if node:
            props = node.get("properties", {})
            results.append({
                "id": node_id,
                "title": props.get("title", "Unknown"),
                "content": props.get("content", "")
            })

    return results, my_community


def build_context(query: str, k: int = 3) -> tuple[str, dict]:
    """Build context for answering a query."""
    # Vector search for seed documents
    seed_docs = vector_search(query, k=k)

    context_parts = []
    seen_ids = set()
    retrieval_info = {
        "vector_search": [],
        "graph_traversal": [],
        "community": []
    }

    for doc in seed_docs:
        doc_id = doc["id"]
        props = doc["properties"]
        title = props.get("title", "Unknown")
        content = props.get("content", "")

        if doc_id not in seen_ids:
            seen_ids.add(doc_id)
            context_parts.append(f"## {title}\n{content}")
            retrieval_info["vector_search"].append({
                "title": title,
                "distance": doc["distance"]
            })

        # Graph traversal
        related_docs = get_related_documents(doc_id)
        for related in related_docs:
            rel_id = related["id"]
            if rel_id and rel_id not in seen_ids:
                seen_ids.add(rel_id)
                rel_title = related.get("title", "Unknown")
                rel_content = related.get("content", "")
                context_parts.append(f"## {rel_title} (via shared entities)\n{rel_content}")
                retrieval_info["graph_traversal"].append({"title": rel_title})

        # Community-based retrieval
        community_docs, community_id = get_community_documents(doc_id, limit=2)
        if community_id != -1:
            for comm_doc in community_docs:
                comm_id = comm_doc["id"]
                if comm_id and comm_id not in seen_ids:
                    seen_ids.add(comm_id)
                    comm_title = comm_doc.get("title", "Unknown")
                    comm_content = comm_doc.get("content", "")
                    context_parts.append(f"## {comm_title} (community {community_id})\n{comm_content}")
                    retrieval_info["community"].append({
                        "title": comm_title,
                        "community_id": community_id
                    })

    return "\n\n".join(context_parts), retrieval_info


def query_graphrag(question: str, use_llm: bool = True) -> dict:
    """
    Answer a question using GraphRAG.

    Args:
        question: The question to answer
        use_llm: Whether to use an LLM for the final answer

    Returns:
        Dict with question, context, retrieval info, and answer
    """
    # Build context
    context, retrieval_info = build_context(question)

    result = {
        "question": question,
        "context": context,
        "retrieval_info": retrieval_info,
        "answer": None
    }

    # LLM inference
    if use_llm and context:
        user_prompt = f"""Context:
{context}

Question: {question}

Answer:"""

        try:
            messages = [
                Message(role="system", content=SYSTEM_PROMPT),
                Message(role="user", content=user_prompt)
            ]
            result["answer"] = ollama_client.chat(messages, temperature=0.3)
        except Exception as e:
            result["answer"] = f"[LLM error: {e}]"
            logger.error(f"LLM error: {e}")

    return result


def get_stats() -> dict:
    """Get knowledge graph statistics."""
    try:
        g = get_graph()

        doc_result = g.connection.cypher("MATCH (n:Document) RETURN count(n) AS cnt")
        doc_count = int(doc_result[0]["cnt"]) if doc_result else 0

        entity_result = g.connection.cypher("MATCH (n:Entity) RETURN count(n) AS cnt")
        entity_count = int(entity_result[0]["cnt"]) if entity_result else 0

        edge_result = g.connection.cypher("MATCH ()-[r:MENTIONS]->() RETURN count(r) AS cnt")
        edge_count = int(edge_result[0]["cnt"]) if edge_result else 0

        return {
            "documents": doc_count,
            "entities": entity_count,
            "relationships": edge_count
        }
    except Exception as e:
        logger.error(f"Error getting stats: {e}")
        return {"documents": 0, "entities": 0, "relationships": 0}


# Flask routes

@app.route('/')
def index():
    """Serve the main page."""
    return render_template('index.html')


@app.route('/static/<path:path>')
def send_static(path):
    """Serve static files."""
    return send_from_directory('static', path)


@app.route('/api/health', methods=['GET'])
def health():
    """Health check endpoint."""
    ollama_available = ollama_client.is_available() if ollama_client else False
    return jsonify({
        "status": "healthy",
        "ollama_available": ollama_available,
        "database": DB_PATH
    })


@app.route('/api/stats', methods=['GET'])
def stats():
    """Get knowledge graph statistics."""
    stats = get_stats()
    return jsonify(stats)


@app.route('/api/ingest', methods=['POST'])
def ingest():
    """Ingest a document into the knowledge graph."""
    data = request.get_json()
    title = data.get('title', '').strip()
    content = data.get('content', '').strip()

    if not title or not content:
        return jsonify({"error": "Title and content are required"}), 400

    result = ingest_document(title, content)
    return jsonify(result)


@app.route('/api/ingest/file', methods=['POST'])
def ingest_file():
    """Ingest a text file into the knowledge graph."""
    if 'file' not in request.files:
        return jsonify({"error": "No file provided"}), 400

    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "No file selected"}), 400

    if not file.filename.endswith('.txt'):
        return jsonify({"error": "Only .txt files are supported"}), 400

    try:
        content = file.read().decode('utf-8')
        title = file.filename.replace('.txt', '')

        result = ingest_document(title, content)
        return jsonify(result)
    except Exception as e:
        logger.error(f"Error reading file: {e}")
        return jsonify({"error": f"Error reading file: {str(e)}"}), 500


@app.route('/api/query', methods=['POST'])
def query():
    """Query the knowledge graph."""
    data = request.get_json()
    question = data.get('question', '').strip()

    if not question:
        return jsonify({"error": "Question is required"}), 400

    use_llm = data.get('use_llm', True)
    result = query_graphrag(question, use_llm=use_llm)

    return jsonify(result)


@app.route('/api/graph', methods=['GET'])
def get_graph_data():
    """Export the knowledge graph structure for visualization."""
    try:
        g = get_graph()

        # Get all documents
        doc_query = "MATCH (n:Document) RETURN n.id AS id, n.title AS title, n.content AS content"
        doc_result = g.connection.cypher(doc_query)

        # Get all entities
        entity_query = "MATCH (n:Entity) RETURN n.id AS id, n.name AS name"
        entity_result = g.connection.cypher(entity_query)

        # Get all relationships
        edge_query = "MATCH (d:Document)-[r:MENTIONS]->(e:Entity) RETURN d.id AS source, e.id AS target"
        edge_result = g.connection.cypher(edge_query)

        # Get communities
        communities = get_communities()

        # Build nodes array
        nodes = []
        for doc in doc_result:
            doc_id = doc.get("id")
            nodes.append({
                "data": {
                    "id": doc_id,
                    "label": doc.get("title", "Unknown"),
                    "type": "document",
                    "content": doc.get("content", "")[:200],  # Truncate for display
                    "community": communities.get(doc_id, -1)
                }
            })

        for entity in entity_result:
            entity_id = entity.get("id")
            nodes.append({
                "data": {
                    "id": entity_id,
                    "label": entity.get("name", "Unknown"),
                    "type": "entity",
                    "community": communities.get(entity_id, -1)
                }
            })

        # Build edges array
        edges = []
        for edge in edge_result:
            edges.append({
                "data": {
                    "source": edge.get("source"),
                    "target": edge.get("target"),
                    "label": "MENTIONS"
                }
            })

        return jsonify({
            "nodes": nodes,
            "edges": edges,
            "stats": {
                "documents": len([n for n in nodes if n["data"]["type"] == "document"]),
                "entities": len([n for n in nodes if n["data"]["type"] == "entity"]),
                "relationships": len(edges),
                "communities": len(set(communities.values()))
            }
        })

    except Exception as e:
        logger.error(f"Error exporting graph: {e}")
        return jsonify({"error": str(e)}), 500


def initialize_app():
    """Initialize the application."""
    global embed_model, ollama_client

    logger.info("Initializing GraphRAG application...")

    # Create directories
    Path("input").mkdir(exist_ok=True)
    Path("output").mkdir(exist_ok=True)
    Path("templates").mkdir(exist_ok=True)
    Path("static").mkdir(exist_ok=True)

    # Initialize thread-safe shared resources
    embed_model = SentenceTransformer("all-MiniLM-L6-v2")
    ollama_client = OllamaClient(model=OLLAMA_MODEL)

    # Initialize database (create tables if needed)
    g = graph(DB_PATH)
    load_vec_extension(g)
    g.close()

    # Check Ollama availability
    if not ollama_client.is_available():
        logger.warning("Ollama is not available. Make sure Ollama is running.")
    else:
        logger.info("Ollama is available")

    logger.info("Application initialized successfully")


if __name__ == '__main__':
    initialize_app()

    # Run Flask app
    port = 8080
    logger.info(f"Starting Flask server on http://localhost:{port}")
    app.run(host='0.0.0.0', port=port, debug=False)

# Made with Bob

The Command Center: `“index.html"`

While the backend does the heavy lifting, the index.html file provides the "cockpit" for the user. It’s built with a clean, modular design using Tailwind CSS for styling and Vis.js for the interactive graph.

Workspace vs. Knowledge Graph: The UI is split into two main views. The Workspace is where the action happens — uploading files and asking questions. The Knowledge Graph tab is where the “invisible” data becomes visible, rendering nodes and edges in real-time.
Dynamic Feedback: The HTML includes dedicated zones for “Retrieval Information” (showing which documents were found) and “Context Retrieved” (showing the exact snippets used to form an answer). This transparency is crucial for verifying AI responses.
Interactive Visualization: Using a specialized canvas, the HTML allows you to zoom, drag, and reorganize the graph. It even includes control buttons to “Fit to Screen” or “Change Layout,” making it easy to explore even the most complex clusters of information.

Conclusion

By combining a robust Python backend with a dynamic HTML frontend, Bob has created more than just a chat interface; he’s built a Knowledge Explorer. It’s a tool that respects the structure of your information, allowing you to see not just what the data says, but how every piece of it is connected.

To make this possible, we (Bob and I 😉) relied on the exceptional foundation provided by the GraphQLite repository. Kudos to the Colliery team who has delivered a masterclass in open-source documentation; the repository is remarkably clean, and the “plug-and-play” nature of their samples made it incredibly easy to move from a fresh clone to a working implementation. By providing a native Cypher query support on top of SQLite, they’ve removed the typical “infrastructure tax” associated with graph databases. It’s rare to find a project that balances such high-level technical capability with such an approachable, developer-friendly entry point — it truly empowered Bob to focus on building the interface rather than fighting the backend.

>>> Thanks for reading <<<

Building an Agent with MCP for Docling XBRL Document Conversion (by Bob)

Alain Airom (Ayrom) — Mon, 23 Mar 2026 08:32:17 +0000

Taming Financial Data: How Bob Built an MCP Agent for Seamless XBRL Conversion provided by Docling!

TL;DR-What is XBRL?

XBRL (eXtensible Business Reporting Language) is a freely available, global standard for exchanging business information. Often described as “barcoding for financial statements,” it uses XML-based tags to identify every individual piece of financial data — such as net profit, total assets, or depreciation — within a digital document. Unlike a static PDF or Excel sheet where data is just text on a page, XBRL makes information machine-readable. This allows software to automatically recognize, select, and analyze specific data points without manual entry, drastically reducing errors and increasing the speed of financial analysis.

XBRL is primarily used by publicly traded companies, government regulators, and financial analysts. In the United States, the SEC (Securities and Exchange Commission) requires companies to file their quarterly and annual reports in XBRL format to ensure transparency and comparability across the market. Similarly, tax authorities, central banks, and stock exchanges worldwide use it to collect and process massive datasets efficiently. By providing a common “language” for reporting, XBRL enables investors and auditors to compare the performance of different companies across various industries and borders with a high degree of precision.

The Docling XBRL Processing Pipeline

To transform complex financial filings into something an AI can actually understand, the agent follows a strict offline-first protocol. This ensures that sensitive financial data never leaves the local environment during the conversion process.

Configuring the Offline Backend

The secret to Docling’s precision is the Taxonomy. Since XBRL documents rely on external schemas to define financial concepts (like “Net Income” or “Total Liabilities”), the agent must be configured to handle these resources locally:

Local Resource Fetching: We set enable_local_fetch to True. This allows the converter to "read" the supporting taxonomy files Bob has stored on the disk.
The Taxonomy Package: To achieve 100% offline operation, Bob provides a Taxonomy Package — a specialized bundle that maps web URLs to local files. Without this, the agent would need to “call home” to regulatory servers to understand the tags.
Remote Fallback: If a package isn’t available, setting enable_remote_fetch to True allows Docling to download and cache these definitions once, then reuse them for future conversions.

The Conversion Lifecycle

Once the environment is set, the agent triggers the convert() method. This isn't just a file format change; it is a full intellectual "ingestion" of the report:

1-Parsing: The agent deconstructs the XBRL instance file (the .xml).

2-Validation: It cross-references the data against the local taxonomy to ensure every financial fact is technically accurate and properly “labeled.”

3-Extraction: It separates the document into three distinct streams:

Metadata: Filing dates, entity names, and reporting periods.
Text Blocks: The narrative “notes” and disclosures often found in 10-Ks.
Numeric Facts: The raw dollars and cents, structured as high-fidelity key-value pairs. 4-Unification: Finally, all these streams are merged into a single DoclingDocument representation.

How parsing (for example with an agent) to read XBRL documents can help?

Because Docling creates this “Unified Representation,” a Model Context Protocol (MCP) can now pass a clean, structured JSON or Markdown version of the financial report to the LLM.

Instead of the LLM guessing what a tag like <us-gaap:CashAndCashEquivalents> means, it receives a clear, human-readable label: "Cash and Cash Equivalents: $557,296." This drastically reduces "hallucinations" and makes financial analysis pinpoint accurate.

Docling provides an example notebook which could be used out-of-the box;

%pip install -q docling
import urllib.request
from pathlib import Path

# Create directories for XBRL data
data_dir = Path("xbrl_data")
taxonomy_dir = data_dir / "taxonomy"
taxonomy_dir.mkdir(parents=True, exist_ok=True)

# Base URL for test data
base_url = (
    "https://raw.githubusercontent.com/docling-project/docling/main/tests/data/xbrl/"
)

# Download XBRL instance file
instance_file = data_dir / "mlac-20251231.xml"
if not instance_file.exists():
    print("Downloading XBRL instance file...")
    urllib.request.urlretrieve(f"{base_url}mlac-20251231.xml", instance_file)
    print(f"Downloaded: {instance_file}")

# Download taxonomy files
taxonomy_files = [
    "mlac-20251231.xsd",
    "mlac-20251231_cal.xml",
    "mlac-20251231_def.xml",
    "mlac-20251231_lab.xml",
    "mlac-20251231_pre.xml",
]

print("Downloading taxonomy files...")
for filename in taxonomy_files:
    target_file = taxonomy_dir / filename
    if not target_file.exists():
        urllib.request.urlretrieve(f"{base_url}mlac-taxonomy/{filename}", target_file)
        print(f"  Downloaded: {filename}")

# Download taxonomy package (contains URL mappings for offline parsing)
taxonomy_package = taxonomy_dir / "taxonomy_package.zip"
if not taxonomy_package.exists():
    print("Downloading taxonomy package...")
    urllib.request.urlretrieve(
        f"{base_url}mlac-taxonomy/taxonomy_package.zip", taxonomy_package
    )
    print("  Downloaded: taxonomy_package.zip")

print("\nAll files downloaded successfully!")
from docling.datamodel.backend_options import XBRLBackendOptions
from docling.datamodel.base_models import InputFormat
from docling.document_converter import DocumentConverter, XBRLFormatOption

# Configure XBRL backend options
backend_options = XBRLBackendOptions(
    enable_local_fetch=True,  # Allow reading local taxonomy files
    enable_remote_fetch=False,  # Disable remote fetching for offline operation
    taxonomy=taxonomy_dir,  # Path to local taxonomy directory
)

# Create document converter with XBRL support
converter = DocumentConverter(
    allowed_formats=[InputFormat.XML_XBRL],
    format_options={
        InputFormat.XML_XBRL: XBRLFormatOption(backend_options=backend_options)
    },
)

print("XBRL converter configured successfully!")

# Convert the XBRL document
print(f"Converting XBRL document: {instance_file}")
result = converter.convert(instance_file)
doc = result.document

print("\nConversion successful!")
print(f"Document name: {doc.name}")
print(f"Number of items: {len(list(doc.iterate_items()))}")

from docling_core.types.doc import DocItemLabel

# Count items by type
item_counts = {}
for item, _ in doc.iterate_items():
    label = item.label
    item_counts[label] = item_counts.get(label, 0) + 1

print("Document structure:")
for label, count in sorted(item_counts.items(), key=lambda x: x[1], reverse=True):
    print(f"  {label.value}: {count}")

# Display first few text items
print("Sample text content:\n")
text_count = 0
for item, _ in doc.iterate_items():
    if item.label == DocItemLabel.TEXT and text_count < 3:
        print(f"- {item.text[:200]}..." if len(item.text) > 200 else f"- {item.text}")
        print()
        text_count += 1

# Display sample key-value pairs
graph_data = doc.key_value_items[0].graph
print(f"Total key-value pairs extracted: {len(graph_data.links)}\n")
for link in graph_data.links[:10]:
    source = next(
        item for item in graph_data.cells if item.cell_id == link.source_cell_id
    )
    target = next(
        item for item in graph_data.cells if item.cell_id == link.target_cell_id
    )
    print(f"{source.text} -> {target.text}")

# Export to Markdown
markdown_content = doc.export_to_markdown()

# Display first 2000 characters
print("Markdown export (first 2000 characters):\n")
print(markdown_content[:2000])
print("\n...")

# Save to file
output_md = data_dir / "output.md"
output_md.write_text(markdown_content)
print(f"\nFull markdown saved to: {output_md}")

import json

# Export to JSON
output_json = data_dir / "output.json"
doc.save_as_json(output_json)

print(f"Document exported to JSON: {output_json}")
print(f"File size: {output_json.stat().st_size / 1024:.2f} KB")

What tool did Bob provide to automate the task?

The Agentic Workflow: From Raw Data to Decision-Ready Insights

Bob’s XBRL Document Conversion Agent operates as a specialized “knowledge worker” within a larger AI ecosystem. By leveraging the Model Context Protocol (MCP), the agent doesn’t just perform a task; it provides a live, structured “context window” for Large Language Models (LLMs) to interact with financial reality. Here is the four-stage process that defines its agentic behavior:

Autonomous Resource Orchestration: Upon activation, the agent independently manages its dependencies. It doesn’t just wait for files; it ensures the environment is “audit-ready” by enabling local resource fetching. By prioritizing a Taxonomy Package, the agent intelligently maps abstract web-based URLs to local definitions, allowing it to function in high-security, offline environments where data privacy is paramount.

"""
XBRL Document Conversion Agent

This module provides a comprehensive agent for converting XBRL (eXtensible Business 
Reporting Language) documents using Docling, with support for offline processing,
taxonomy validation, and multiple export formats.

XBRL is a standard XML-based format used globally by companies, regulators, and 
financial institutions for exchanging business and financial information in a 
structured, machine-readable format.

Author: Generated for XBRL Document Conversion
License: MIT
"""

import logging
from pathlib import Path
from typing import Optional, Dict, List, Any, Union
from dataclasses import dataclass, field
import json

try:
    from docling.document_converter import DocumentConverter, PdfFormatOption, XBRLFormatOption
    from docling.datamodel.base_models import InputFormat
    from docling.datamodel.pipeline_options import PipelineOptions
    from docling.datamodel.backend_options import XBRLBackendOptions
    from docling.datamodel.document import ConversionResult, DoclingDocument
    DOCLING_AVAILABLE = True
except ImportError as e:
    DOCLING_AVAILABLE = False
    # Define placeholder types when docling is not available
    ConversionResult = Any
    DoclingDocument = Any
    logging.warning(f"Docling library not available. Install with: pip install docling[xbrl]. Error: {e}")


# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


@dataclass
class XBRLConversionConfig:
    """
    Configuration for XBRL document conversion.

    Attributes:
        enable_local_fetch: Enable fetching taxonomy files from local directory
        enable_remote_fetch: Enable fetching taxonomy files from remote URLs
        taxonomy_dir: Path to local taxonomy directory containing schema and linkbase files
        taxonomy_package: Path to taxonomy package ZIP file for offline operation
        output_dir: Directory for saving converted documents
        export_formats: List of export formats (markdown, json, html, text)
    """
    enable_local_fetch: bool = True
    enable_remote_fetch: bool = False
    taxonomy_dir: Optional[Path] = None
    taxonomy_package: Optional[Path] = None
    output_dir: Path = Path("./output")
    export_formats: List[str] = field(default_factory=lambda: ["markdown", "json"])

    def __post_init__(self):
        """Validate configuration after initialization."""
        if self.enable_local_fetch and not self.taxonomy_dir:
            logger.warning("Local fetch enabled but no taxonomy_dir specified")

        if not self.enable_local_fetch and not self.enable_remote_fetch:
            raise ValueError("At least one of enable_local_fetch or enable_remote_fetch must be True")

        # Ensure output directory exists
        self.output_dir.mkdir(parents=True, exist_ok=True)


class XBRLConversionAgent:
    """
    Agent for converting XBRL documents to various formats.

    This agent handles:
    - XBRL instance document parsing
    - Taxonomy validation (local or remote)
    - Metadata extraction
    - Text block extraction
    - Numeric fact extraction
    - Export to multiple formats (Markdown, JSON, HTML, etc.)

    Example:
        >>> config = XBRLConversionConfig(
        ...     taxonomy_dir=Path("./taxonomy"),
        ...     taxonomy_package=Path("./taxonomy/package.zip")
        ... )
        >>> agent = XBRLConversionAgent(config)
        >>> result = agent.convert_document("report.xml")
        >>> agent.export_document(result.document, "output")
    """

    def __init__(self, config: XBRLConversionConfig):
        """
        Initialize the XBRL conversion agent.

        Args:
            config: Configuration object for XBRL conversion

        Raises:
            ImportError: If Docling library is not available
            ValueError: If configuration is invalid
        """
        if not DOCLING_AVAILABLE:
            raise ImportError(
                "Docling library is required. Install with: pip install docling[xbrl]"
            )

        self.config = config
        self.converter = self._setup_converter()
        logger.info("XBRL Conversion Agent initialized successfully")

    def _setup_converter(self) -> DocumentConverter:
        """
        Set up the DocumentConverter with XBRL backend configuration.

        Returns:
            Configured DocumentConverter instance
        """
        # Configure XBRL backend options
        # Use taxonomy_package if available, otherwise use taxonomy_dir
        taxonomy_path = None
        if self.config.taxonomy_package and self.config.taxonomy_package.exists():
            taxonomy_path = self.config.taxonomy_package
        elif self.config.taxonomy_dir and self.config.taxonomy_dir.exists():
            taxonomy_path = self.config.taxonomy_dir

        # Create backend options
        backend_options = XBRLBackendOptions(
            enable_local_fetch=self.config.enable_local_fetch,
            enable_remote_fetch=self.config.enable_remote_fetch,
            taxonomy=taxonomy_path
        )

        # Create format options with XBRLFormatOption wrapper
        # This is the correct way according to Docling's test suite
        format_options = {
            InputFormat.XML_XBRL: XBRLFormatOption(backend_options=backend_options)
        }

        # Create and return converter
        converter = DocumentConverter(
            allowed_formats=[InputFormat.XML_XBRL],
            format_options=format_options
        )

        logger.info("XBRL converter configured successfully")
        return converter

    def convert_document(
        self, 
        xbrl_path: Union[str, Path]
    ) -> ConversionResult:
        """
        Convert an XBRL instance document to DoclingDocument format.

        This method:
        1. Parses the XBRL instance file
        2. Validates against the taxonomy
        3. Extracts metadata, text blocks, and numeric facts
        4. Returns a unified DoclingDocument representation

        Args:
            xbrl_path: Path to XBRL instance document (.xml file)

        Returns:
            ConversionResult containing the converted document and metadata

        Raises:
            FileNotFoundError: If XBRL file doesn't exist
            ValueError: If conversion fails
        """
        xbrl_path = Path(xbrl_path)

        if not xbrl_path.exists():
            raise FileNotFoundError(f"XBRL file not found: {xbrl_path}")

        logger.info(f"Converting XBRL document: {xbrl_path}")

        try:
            result = self.converter.convert(xbrl_path)

            if result.status.name != "SUCCESS":
                raise ValueError(f"Conversion failed with status: {result.status}")

            logger.info(f"Conversion successful! Document: {result.document.name}")
            logger.info(f"Number of items: {len(list(result.document.iterate_items()))}")

            return result

        except Exception as e:
            logger.error(f"Error converting XBRL document: {e}")
            raise

    def get_document_structure(self, document: DoclingDocument) -> Dict[str, int]:
        """
        Analyze and return the structure of a converted document.

        Args:
            document: Converted DoclingDocument

        Returns:
            Dictionary mapping item types to their counts
        """
        structure = {}

        for item, _ in document.iterate_items():
            item_type = type(item).__name__
            structure[item_type] = structure.get(item_type, 0) + 1

        logger.info(f"Document structure: {structure}")
        return structure

    def extract_key_value_pairs(self, document: DoclingDocument) -> List[Dict[str, Any]]:
        """
        Extract numeric facts as key-value pairs from XBRL document.

        XBRL numeric facts are represented as key-value pairs in the document.
        This method extracts all such pairs for analysis.

        Args:
            document: Converted DoclingDocument

        Returns:
            List of dictionaries containing key-value pairs with metadata
        """
        key_value_pairs = []

        # Extract from key-value regions
        for kv_item in document.key_value_items:
            key_value_pairs.append({
                "key": kv_item.label if hasattr(kv_item, 'label') else str(kv_item),
                "value": kv_item.text if hasattr(kv_item, 'text') else None,
                "type": "key_value_region"
            })

        logger.info(f"Extracted {len(key_value_pairs)} key-value pairs")
        return key_value_pairs

    def extract_text_content(
        self, 
        document: DoclingDocument, 
        max_items: int = 10
    ) -> List[str]:
        """
        Extract sample text content from the document.

        Args:
            document: Converted DoclingDocument
            max_items: Maximum number of text items to extract

        Returns:
            List of text strings
        """
        text_items = []

        for item in document.texts[:max_items]:
            if hasattr(item, 'text') and item.text:
                # Truncate long text for readability
                text = item.text[:200] + "..." if len(item.text) > 200 else item.text
                text_items.append(text)

        logger.info(f"Extracted {len(text_items)} text items")
        return text_items

    def export_to_markdown(
        self, 
        document: DoclingDocument, 
        output_path: Optional[Path] = None
    ) -> str:
        """
        Export document to Markdown format.

        Args:
            document: Converted DoclingDocument
            output_path: Optional path to save the markdown file

        Returns:
            Markdown string
        """
        markdown = document.export_to_markdown()

        if output_path:
            output_path = Path(output_path)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(markdown, encoding='utf-8')
            logger.info(f"Markdown exported to: {output_path}")

        return markdown

    def export_to_json(
        self,
        document: DoclingDocument,
        output_path: Optional[Path] = None
    ) -> str:
        """
        Export document to JSON format.

        Args:
            document: Converted DoclingDocument
            output_path: Optional path to save the JSON file

        Returns:
            JSON string
        """
        # Use the document's dict representation and convert to JSON
        json_str = json.dumps(document.model_dump(), indent=2, ensure_ascii=False)

        if output_path:
            output_path = Path(output_path)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(json_str, encoding='utf-8')

            # Calculate file size
            size_kb = len(json_str.encode('utf-8')) / 1024
            logger.info(f"JSON exported to: {output_path} ({size_kb:.2f} KB)")

        return json_str

    def export_to_html(
        self, 
        document: DoclingDocument, 
        output_path: Optional[Path] = None
    ) -> str:
        """
        Export document to HTML format.

        Args:
            document: Converted DoclingDocument
            output_path: Optional path to save the HTML file

        Returns:
            HTML string
        """
        html = document.export_to_html()

        if output_path:
            output_path = Path(output_path)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(html, encoding='utf-8')
            logger.info(f"HTML exported to: {output_path}")

        return html

    def export_document(
        self, 
        document: DoclingDocument, 
        base_name: str,
        formats: Optional[List[str]] = None
    ) -> Dict[str, Path]:
        """
        Export document to multiple formats.

        Args:
            document: Converted DoclingDocument
            base_name: Base name for output files (without extension)
            formats: List of formats to export (default: from config)

        Returns:
            Dictionary mapping format names to output file paths
        """
        formats = formats or self.config.export_formats
        output_files = {}

        for fmt in formats:
            output_path = self.config.output_dir / f"{base_name}.{fmt}"

            if fmt == "markdown" or fmt == "md":
                self.export_to_markdown(document, output_path)
                output_files["markdown"] = output_path

            elif fmt == "json":
                self.export_to_json(document, output_path)
                output_files["json"] = output_path

            elif fmt == "html":
                self.export_to_html(document, output_path)
                output_files["html"] = output_path

            else:
                logger.warning(f"Unsupported export format: {fmt}")

        return output_files

    def process_xbrl_file(
        self, 
        xbrl_path: Union[str, Path],
        output_base_name: Optional[str] = None,
        analyze: bool = True
    ) -> Dict[str, Any]:
        """
        Complete processing pipeline for an XBRL file.

        This is a convenience method that:
        1. Converts the XBRL document
        2. Analyzes the structure (optional)
        3. Exports to configured formats

        Args:
            xbrl_path: Path to XBRL instance document
            output_base_name: Base name for output files (default: input filename)
            analyze: Whether to perform structure analysis

        Returns:
            Dictionary containing conversion results and analysis
        """
        xbrl_path = Path(xbrl_path)

        if not output_base_name:
            output_base_name = xbrl_path.stem

        # Convert document
        result = self.convert_document(xbrl_path)
        document = result.document

        # Prepare results
        results = {
            "input_file": str(xbrl_path),
            "document_name": document.name,
            "conversion_status": result.status.name,
        }

        # Analyze structure if requested
        if analyze:
            results["structure"] = self.get_document_structure(document)
            results["key_value_pairs"] = self.extract_key_value_pairs(document)
            results["sample_text"] = self.extract_text_content(document, max_items=5)

        # Export to configured formats
        output_files = self.export_document(document, output_base_name)
        results["output_files"] = {k: str(v) for k, v in output_files.items()}

        logger.info(f"Processing complete for {xbrl_path}")
        return results


def create_agent_from_taxonomy(
    taxonomy_dir: Union[str, Path],
    taxonomy_package: Optional[Union[str, Path]] = None,
    output_dir: Union[str, Path] = "./output",
    enable_remote_fetch: bool = False
) -> XBRLConversionAgent:
    """
    Factory function to create an XBRL agent with taxonomy configuration.

    Args:
        taxonomy_dir: Path to directory containing taxonomy files
        taxonomy_package: Optional path to taxonomy package ZIP
        output_dir: Directory for output files
        enable_remote_fetch: Whether to allow remote taxonomy fetching

    Returns:
        Configured XBRLConversionAgent instance

    Example:
        >>> agent = create_agent_from_taxonomy(
        ...     taxonomy_dir="./data/xbrl/mlac-taxonomy",
        ...     taxonomy_package="./data/xbrl/mlac-taxonomy/taxonomy_package.zip"
        ... )
        >>> result = agent.process_xbrl_file("./data/xbrl/mlac-20251231.xml")
    """
    config = XBRLConversionConfig(
        enable_local_fetch=True,
        enable_remote_fetch=enable_remote_fetch,
        taxonomy_dir=Path(taxonomy_dir),
        taxonomy_package=Path(taxonomy_package) if taxonomy_package else None,
        output_dir=Path(output_dir)
    )

    return XBRLConversionAgent(config)


if __name__ == "__main__":
    # Example usage
    print("XBRL Document Conversion Agent")
    print("=" * 50)
    print("\nThis module provides tools for converting XBRL documents.")
    print("Import and use the XBRLConversionAgent class in your code.")
    print("\nExample:")
    print("  from xbrl_agent import create_agent_from_taxonomy")
    print("  agent = create_agent_from_taxonomy('./taxonomy')")
    print("  result = agent.process_xbrl_file('report.xml')")

# Made with Bob

Intelligent Validation & Parsing: When an XBRL instance document is submitted, the agent acts as a digital auditor. It doesn’t merely “read” the XML; it validates the data against the specific financial taxonomy. Using the Docling backend, it deconstructs the file into three critical streams — metadata, narrative text blocks, and numeric facts — ensuring that every data point is contextually accurate before it ever reaches the user.
Unified Context Synthesis: The agent’s “brain” (the Docling Converter) takes these disparate financial streams and synthesizes them into a Unified DoclingDocument. This stage is crucial: it transforms raw, nested XML tags into a clean, hierarchical structure. By flattening complex numeric facts into machine-readable key-value pairs, the agent removes the technical “noise” that typically causes AI models to hallucinate during financial analysis.
MCP-Enabled Interactivity: Finally, the agent exposes this processed intelligence via an MCP Server. This is where the magic happens: instead of the LLM receiving a massive, unreadable wall of XML, the agent serves up a precise “Context Retrieval” feed. Whether through a Web UI or a direct model connection, the agent stands ready to answer natural language queries like “What was the net change in cash?” by pulling exactly the right structured facts from its memory.

The MCP server provides 6 tools for XBRL conversion:

xbrl_create_agent — Create conversion agent with taxonomy
xbrl_convert_document — Convert XBRL document
xbrl_analyze_structure — Analyze document structure
xbrl_extract_key_values — Extract numeric facts
xbrl_extract_text — Extract text content
xbrl_export_document — Export to formats

#!/usr/bin/env python3
"""
XBRL MCP Server

Model Context Protocol (MCP) server for XBRL document conversion.
This server exposes XBRL conversion capabilities as MCP tools that can be
used by AI assistants and other MCP clients.

Usage:
    python xbrl_mcp_server.py

Or as an MCP server in your MCP configuration:
    {
      "mcpServers": {
        "xbrl": {
          "command": "python",
          "args": ["/path/to/xbrl_mcp_server.py"]
        }
      }
    }
"""

import asyncio
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

try:
    from mcp.server import Server
    from mcp.server.stdio import stdio_server
    from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
    MCP_AVAILABLE = True
except ImportError:
    MCP_AVAILABLE = False
    logging.warning("MCP library not available. Install with: pip install mcp")

from xbrl_agent import (
    XBRLConversionAgent,
    XBRLConversionConfig,
    create_agent_from_taxonomy,
    DOCLING_AVAILABLE
)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class XBRLMCPServer:
    """
    MCP Server for XBRL document conversion.

    Provides tools for:
    - Converting XBRL documents
    - Analyzing XBRL structure
    - Extracting data from XBRL
    - Exporting to various formats
    """

    def __init__(self):
        """Initialize the XBRL MCP server."""
        if not MCP_AVAILABLE:
            raise ImportError("MCP library required. Install with: pip install mcp")

        if not DOCLING_AVAILABLE:
            raise ImportError("Docling library required. Install with: pip install docling[xbrl]")

        self.server = Server("xbrl-converter")
        self.agents: Dict[str, XBRLConversionAgent] = {}
        self._setup_tools()
        logger.info("XBRL MCP Server initialized")

    def _setup_tools(self):
        """Register all MCP tools."""

        @self.server.list_tools()
        async def list_tools() -> List[Tool]:
            """List all available XBRL conversion tools."""
            return [
                Tool(
                    name="xbrl_create_agent",
                    description=(
                        "Create an XBRL conversion agent with taxonomy configuration. "
                        "This must be called before using other XBRL tools. "
                        "Returns an agent_id to use in subsequent calls."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Unique identifier for this agent"
                            },
                            "taxonomy_dir": {
                                "type": "string",
                                "description": "Path to directory containing taxonomy files"
                            },
                            "taxonomy_package": {
                                "type": "string",
                                "description": "Optional path to taxonomy package ZIP file"
                            },
                            "output_dir": {
                                "type": "string",
                                "description": "Directory for output files (default: ./output)",
                                "default": "./output"
                            },
                            "enable_remote_fetch": {
                                "type": "boolean",
                                "description": "Allow fetching taxonomy from remote URLs",
                                "default": False
                            }
                        },
                        "required": ["agent_id", "taxonomy_dir"]
                    }
                ),
                Tool(
                    name="xbrl_convert_document",
                    description=(
                        "Convert an XBRL instance document to DoclingDocument format. "
                        "Parses the XBRL file, validates against taxonomy, and extracts "
                        "metadata, text blocks, and numeric facts."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Agent ID from xbrl_create_agent"
                            },
                            "xbrl_path": {
                                "type": "string",
                                "description": "Path to XBRL instance document (.xml file)"
                            },
                            "output_base_name": {
                                "type": "string",
                                "description": "Base name for output files (optional)"
                            },
                            "analyze": {
                                "type": "boolean",
                                "description": "Perform structure analysis",
                                "default": True
                            }
                        },
                        "required": ["agent_id", "xbrl_path"]
                    }
                ),
                Tool(
                    name="xbrl_analyze_structure",
                    description=(
                        "Analyze the structure of a converted XBRL document. "
                        "Returns counts of different item types (text, tables, etc.)."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Agent ID from xbrl_create_agent"
                            },
                            "xbrl_path": {
                                "type": "string",
                                "description": "Path to XBRL instance document"
                            }
                        },
                        "required": ["agent_id", "xbrl_path"]
                    }
                ),
                Tool(
                    name="xbrl_extract_key_values",
                    description=(
                        "Extract numeric facts as key-value pairs from XBRL document. "
                        "Returns all XBRL facts with their values and metadata."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Agent ID from xbrl_create_agent"
                            },
                            "xbrl_path": {
                                "type": "string",
                                "description": "Path to XBRL instance document"
                            },
                            "max_items": {
                                "type": "integer",
                                "description": "Maximum number of items to return",
                                "default": 100
                            }
                        },
                        "required": ["agent_id", "xbrl_path"]
                    }
                ),
                Tool(
                    name="xbrl_extract_text",
                    description=(
                        "Extract text content from XBRL document. "
                        "Returns text blocks and narrative content."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Agent ID from xbrl_create_agent"
                            },
                            "xbrl_path": {
                                "type": "string",
                                "description": "Path to XBRL instance document"
                            },
                            "max_items": {
                                "type": "integer",
                                "description": "Maximum number of text items to return",
                                "default": 10
                            }
                        },
                        "required": ["agent_id", "xbrl_path"]
                    }
                ),
                Tool(
                    name="xbrl_export_document",
                    description=(
                        "Export XBRL document to specified formats. "
                        "Supports markdown, json, and html formats."
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "agent_id": {
                                "type": "string",
                                "description": "Agent ID from xbrl_create_agent"
                            },
                            "xbrl_path": {
                                "type": "string",
                                "description": "Path to XBRL instance document"
                            },
                            "output_base_name": {
                                "type": "string",
                                "description": "Base name for output files"
                            },
                            "formats": {
                                "type": "array",
                                "items": {"type": "string"},
                                "description": "Export formats (markdown, json, html)",
                                "default": ["markdown", "json"]
                            }
                        },
                        "required": ["agent_id", "xbrl_path", "output_base_name"]
                    }
                ),
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: Any) -> List[TextContent]:
            """Handle tool calls."""
            try:
                if name == "xbrl_create_agent":
                    return await self._create_agent(arguments)
                elif name == "xbrl_convert_document":
                    return await self._convert_document(arguments)
                elif name == "xbrl_analyze_structure":
                    return await self._analyze_structure(arguments)
                elif name == "xbrl_extract_key_values":
                    return await self._extract_key_values(arguments)
                elif name == "xbrl_extract_text":
                    return await self._extract_text(arguments)
                elif name == "xbrl_export_document":
                    return await self._export_document(arguments)
                else:
                    return [TextContent(
                        type="text",
                        text=f"Unknown tool: {name}"
                    )]
            except Exception as e:
                logger.error(f"Error in tool {name}: {e}")
                return [TextContent(
                    type="text",
                    text=f"Error: {str(e)}"
                )]

    async def _create_agent(self, args: Dict[str, Any]) -> List[TextContent]:
        """Create an XBRL conversion agent."""
        agent_id = args["agent_id"]

        if agent_id in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' already exists. Use a different agent_id."
            )]

        try:
            agent = create_agent_from_taxonomy(
                taxonomy_dir=args["taxonomy_dir"],
                taxonomy_package=args.get("taxonomy_package"),
                output_dir=args.get("output_dir", "./output"),
                enable_remote_fetch=args.get("enable_remote_fetch", False)
            )

            self.agents[agent_id] = agent

            result = {
                "status": "success",
                "agent_id": agent_id,
                "message": f"XBRL agent '{agent_id}' created successfully",
                "config": {
                    "taxonomy_dir": args["taxonomy_dir"],
                    "output_dir": args.get("output_dir", "./output"),
                    "enable_remote_fetch": args.get("enable_remote_fetch", False)
                }
            }

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def _convert_document(self, args: Dict[str, Any]) -> List[TextContent]:
        """Convert an XBRL document."""
        agent_id = args["agent_id"]

        if agent_id not in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' not found. Create it first with xbrl_create_agent."
            )]

        agent = self.agents[agent_id]

        try:
            result = agent.process_xbrl_file(
                xbrl_path=args["xbrl_path"],
                output_base_name=args.get("output_base_name"),
                analyze=args.get("analyze", True)
            )

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def _analyze_structure(self, args: Dict[str, Any]) -> List[TextContent]:
        """Analyze XBRL document structure."""
        agent_id = args["agent_id"]

        if agent_id not in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' not found."
            )]

        agent = self.agents[agent_id]

        try:
            conv_result = agent.convert_document(args["xbrl_path"])
            structure = agent.get_document_structure(conv_result.document)

            result = {
                "status": "success",
                "document_name": conv_result.document.name,
                "structure": structure,
                "total_items": sum(structure.values())
            }

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def _extract_key_values(self, args: Dict[str, Any]) -> List[TextContent]:
        """Extract key-value pairs from XBRL document."""
        agent_id = args["agent_id"]

        if agent_id not in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' not found."
            )]

        agent = self.agents[agent_id]

        try:
            conv_result = agent.convert_document(args["xbrl_path"])
            kv_pairs = agent.extract_key_value_pairs(conv_result.document)

            max_items = args.get("max_items", 100)
            kv_pairs = kv_pairs[:max_items]

            result = {
                "status": "success",
                "document_name": conv_result.document.name,
                "key_value_pairs": kv_pairs,
                "count": len(kv_pairs)
            }

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def _extract_text(self, args: Dict[str, Any]) -> List[TextContent]:
        """Extract text content from XBRL document."""
        agent_id = args["agent_id"]

        if agent_id not in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' not found."
            )]

        agent = self.agents[agent_id]

        try:
            conv_result = agent.convert_document(args["xbrl_path"])
            max_items = args.get("max_items", 10)
            texts = agent.extract_text_content(conv_result.document, max_items)

            result = {
                "status": "success",
                "document_name": conv_result.document.name,
                "text_items": texts,
                "count": len(texts)
            }

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def _export_document(self, args: Dict[str, Any]) -> List[TextContent]:
        """Export XBRL document to specified formats."""
        agent_id = args["agent_id"]

        if agent_id not in self.agents:
            return [TextContent(
                type="text",
                text=f"Agent '{agent_id}' not found."
            )]

        agent = self.agents[agent_id]

        try:
            conv_result = agent.convert_document(args["xbrl_path"])
            output_files = agent.export_document(
                document=conv_result.document,
                base_name=args["output_base_name"],
                formats=args.get("formats", ["markdown", "json"])
            )

            result = {
                "status": "success",
                "document_name": conv_result.document.name,
                "output_files": {k: str(v) for k, v in output_files.items()}
            }

            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

        except Exception as e:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": "error",
                    "message": str(e)
                }, indent=2)
            )]

    async def run(self):
        """Run the MCP server."""
        async with stdio_server() as (read_stream, write_stream):
            logger.info("XBRL MCP Server running on stdio")
            await self.server.run(
                read_stream,
                write_stream,
                self.server.create_initialization_options()
            )


async def main():
    """Main entry point for the MCP server."""
    try:
        server = XBRLMCPServer()
        await server.run()
    except ImportError as e:
        logger.error(f"Missing dependency: {e}")
        logger.error("Install with: pip install mcp docling[xbrl]")
    except Exception as e:
        logger.error(f"Server error: {e}")
        raise


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

# Made with Bob

Simplified User Interface: Last but not least, there is an intutive UI in order to test and interact with the mechanism and test it easily.

Bob’s XBRL Agent: The MCP Toolset Cheat Sheet

By exposing these six specialized tools through the Model Context Protocol, Bob has given the AI assistant a “financial expert” toolkit. Instead of struggling with XML, the assistant simply calls the appropriate tool to get clean, validated data.

| Tool Name                     | Agentic Responsibility     | Business Value                                               |
| ----------------------------- | -------------------------- | ------------------------------------------------------------ |
| **`xbrl_create_agent`**       | **Resource Orchestration** | Sets up the secure, offline environment by loading specific legal and accounting taxonomies. |
| **`xbrl_convert_document`**   | **Context Synthesis**      | The "heavy lifter" that transforms raw `.xml` into a unified, machine-readable `DoclingDocument`. |
| **`xbrl_analyze_structure`**  | **Structural Auditing**    | Provides a high-level map of the filing (e.g., identifying where the Balance Sheet or Risk Factors are). |
| **`xbrl_extract_key_values`** | **Fact Retrieval**         | Pulls precise numeric facts (Assets, Liabilities, Revenue) as ready-to-use key-value pairs. |
| **`xbrl_extract_text`**       | **Narrative Extraction**   | Isolates the "Management Discussion" and "Footnotes" from the numeric data for qualitative analysis. |
| **`xbrl_export_document`**    | **Output Delivery**        | Finalizes the process by generating clean Markdown or JSON for reports and downstream dashboards. |

Conclusion: The Impact of Agentic XBRL Conversion

By moving from a static conversion script to a Model Context Protocol (MCP) enabled agent, Bob has transformed a high-friction financial task into a seamless, conversational experience. This architecture represents a shift in how we handle specialized data: instead of forcing the AI to learn the intricacies of XBRL, we provide it with a “expert tool” that handles the complexity on its behalf.

For Bob’s organization, the results are clear:

Zero Hallucination Risk: By providing the LLM with structured facts rather than raw XML, the agent ensures that the AI’s answers are grounded in validated financial truths.
Enhanced Security: With local taxonomy fetching and offline processing, sensitive internal or pre-release filings never leave the secure environment.
Plug-and-Play Intelligence: Because it uses the MCP standard, this agent can be plugged into any modern AI assistant (like watsonx Orchestrate, Claude, ChatGPT, or custom internal bots) without rewriting a single line of code.

Ultimately, Bob’s project demonstrates that the future of document processing isn’t just about “reading” files — it’s about building agentic bridges that translate the world’s most complex data into a language both humans and machines can finally agree on.

>>> Thanks for reading <<<

Links and ressources

Blog Post’s Code Repository: https://github.com/aairom/docling-xbrl-test
Docling Project: https://github.com/docling-project
Docling Documentation: https://docling-project.github.io/docling/
Docling XBRL Document Conversion: https://docling-project.github.io/docling/examples/xbrl_conversion/#export-to-json
Docling’s repository “test_backend_xbrl.py”: https://github.com/docling-project/docling/blob/main/tests/test_backend_xbrl.py
Test data for XBRL Conversion: https://github.com/docling-project/docling/tree/main/tests/data/xbrl
IBM Project Bob (GA tomorrow March 24th): https://www.ibm.com/products/bob

Locating the Hidden: Using HashiCorp Vault Radar 📡 to Audit Your Local Machine for Risks

Alain Airom (Ayrom) — Sat, 21 Mar 2026 21:38:54 +0000

Prevent yourself sending sensitive information proactively!

We’ve all been there: you’re deep in the flow of building a solution, confidently hardcoding a temporary API key or database password locally just to get the “thing” working. We tell ourselves we’ll scrub the code before the final push, but in the rush to deploy to the cloud or sync with Git, those “temporary” secrets often slip through the cracks. This false sense of security has led to catastrophic data breaches and exposed infrastructures across the industry. HashiCorp Vault Radar serves as your final line of defense, scanning your environment to detect those hidden vulnerabilities and stopping sensitive “stuff” from ever going public, ensuring your confidence is backed by actual security.

Stopping the Leak: An Introduction to HashiCorp Vault Radar

In the modern DevSecOps landscape, a single “git push” can become a multi-million dollar mistake. As development cycles accelerate, the risk of accidentally embedding sensitive data — passwords, API keys, or private certificates — directly into source code increases exponentially. Once a secret is committed to a repository’s history, it is no longer secret.

What is Vault Radar?

Vault Radar is a specialized security product within the HashiCorp Cloud Platform (HCP) designed to automate the detection and identification of unmanaged secrets. While HashiCorp Vault is famous for managing secrets, Vault Radar is the “lookout” that finds the ones you forgot to protect.

It acts as a continuous scanning engine that looks beyond just active code; it dives deep into your version control history to identify risks that might have been buried years ago.

What Does it Scan For?

Vault Radar isn’t just a regex-based secret hunter. It provides a multi-layered approach to identifying risks:

Secrets: It identifies passwords, tokens, and keys. Crucially, it can distinguish between an inactive string and an active secret, helping teams prioritize the most critical threats.
Personally Identifiable Information (PII): As global privacy regulations (like GDPR and CCPA) tighten, Vault Radar helps identify sensitive user data that shouldn’t be in your codebase.
Non-Inclusive Language (NIL): It helps maintain brand integrity and community standards by flagging non-inclusive terms in your documentation and code.

How It Fits Into Your Workflow

Vault Radar doesn’t just wait for a security audit to run. It integrates directly into the developer lifecycle:

Real-Time Scanning: It scans as you work, checking new commits and pull requests (PRs).
Historical Deep Scans: It analyzes the entire Git history of a repository to find “zombie” secrets.
Prioritized Remediation: Using a sophisticated severity ranking (from Info to Critical), it tells security teams exactly what needs immediate attention based on whether a secret is active and where it is located.

Why Does It Matter?

Traditional secret scanning often results in “alert fatigue” — hundreds of false positives that developers eventually ignore. Vault Radar addresses this by using advanced hashing (like Argon2id) and context-aware rules to ensure that when it flags a risk, it’s a risk worth fixing.

By catching secrets before they reach production — and cleaning up the ones that already did — Vault Radar bridges the gap between development speed and enterprise-grade security.

How to use Vault-Radar Locally on your own Machine?

Vault Radar Local Intallation

Hereafter the procedure I implemented on my local machine.

macOS

# Using Homebrew
brew tap hashicorp/tap
brew install hashicorp/tap/vault-radar

Linux

# Download the binary
wget https://releases.hashicorp.com/vault-radar/0.10.0/vault-radar_0.10.0_linux_amd64.zip

# Unzip
unzip vault-radar_0.10.0_linux_amd64.zip

# Move to PATH
sudo mv vault-radar /usr/local/bin/

# Make executable
sudo chmod +x /usr/local/bin/vault-radar

Verify Installation

vault-radar --version

Step 1: Get Your HCP Project ID

Go to HashiCorp Cloud Platform

Sign in or create a free account
Once logged in, either you create OR you’ll see your organization and projects
Click on your project name
Look at the URL in your browser — it will look like:

https://portal.cloud.hashicorp.com/orgs/YOUR-ORG/projects/YOUR-PROJECT-ID

Copy the YOUR-PROJECT-ID part (it's a UUID like

a1b2c3d4-e5f6-7890-abcd-ef1234567890

Alternative method:

In your project, click the ⚙️ Settings icon in the left sidebar
The Project ID is displayed at the top of the settings page

Create a Service Principal (for Client ID and Secret)

In your HCP project, click Access control (IAM) in the left sidebar
Click the Service principals tab
Click Create service principal
Give it a name (e.g., “vault-radar-scanner”)
Assign it the Contributor role (or a custom role with Vault Radar permissions)
Click Save

🚨 CRITICAL — Client Secret is Only Shown Once! 🚨

After clicking Save, a popup/modal will immediately appear with:
Client ID (you can view this later)
Client Secret (⚠️ ONLY SHOWN NOW — cannot be retrieved later!)

You MUST copy the Client Secret immediately!

┌─────────────────────────────────────────────┐
│  Service Principal Created                  │
├─────────────────────────────────────────────┤
│  Client ID:                                 │
│  abc123def456ghi789jkl012mno345pq          │
│  [Copy]                                     │
│                                             │
│  Client Secret: (shown only once)           │
│  aBcDeFgHiJkLmNoPqRsTuVwXyZ123456...       │
│  [Copy]                                     │
│                                             │
│  ⚠️  Save this secret now. You won't be     │
│     able to see it again.                   │
└─────────────────────────────────────────────┘

Click the [Copy] button next to the Client Secret or manually select and copy the entire secret string.

🫣 What if I closed the popup without copying the secret?

If you missed copying the Client Secret, you have two options:

Option A: Create a New Service Principal

Go back to Access control (IAM) → Service principals
Create a new service principal (give it a different name like “vault-radar-scanner-2”)
This time, copy the Client Secret immediately
Optionally, delete the old service principal you can’t use
Option B: Generate a New Key for Existing Service Principal
Go to Access control (IAM) → Service principals
Click on your existing service principal
Look for Keys or Client credentials section
Click Generate new key or Create key
Copy the new Client Secret immediately (the old one will be invalidated).

Step 3: Set Environment Variables

Method 1: Direct Export (Quick Test)

# Replace with your actual values from steps 1 and 2
export HCP_PROJECT_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"
export HCP_CLIENT_ID="abc123def456ghi789jkl012mno345pq"
export HCP_CLIENT_SECRET="aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890AbCdEfGhIjKlMnOpQrStUvWxYz12"

Method 2: Using a .env File (Recommended)

Create a .env file in your project directory:

# .env file
export HCP_PROJECT_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"
export HCP_CLIENT_ID="abc123def456ghi789jkl012mno345pq"
export HCP_CLIENT_SECRET="aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890AbCdEfGhIjKlMnOpQrStUvWxYz12"

Important: Add .env to your .gitignore to avoid committing secrets:

echo ".env" >> .gitignore

Load the .env file before running vault-radar:

# Source the .env file to load variables into your current shell
source .env

# Or use this shorthand
. .env

# Now run vault-radar
vault-radar scan folder -p . -o results.json -f json

Note: The .env file only affects the current terminal session. You need to run source .env each time you open a new terminal, or add the exports to your shell profile for persistence.

Step 4: Persist Environment Variables (Optional but Recommended)

To avoid setting these every time you open a terminal, add them to your shell profile:

For Bash users:

# Add to ~/.bashrc or ~/.bash_profile
echo 'export HCP_PROJECT_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"' >> ~/.bashrc
echo 'export HCP_CLIENT_ID="abc123def456ghi789jkl012mno345pq"' >> ~/.bashrc
echo 'export HCP_CLIENT_SECRET="aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890AbCdEfGhIjKlMnOpQrStUvWxYz12"' >> ~/.bashrc

# Reload configuration
source ~/.bashrc

For Zsh users (macOS default):

# Add to ~/.zshrc
echo 'export HCP_PROJECT_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"' >> ~/.zshrc
echo 'export HCP_CLIENT_ID="abc123def456ghi789jkl012mno345pq"' >> ~/.zshrc
echo 'export HCP_CLIENT_SECRET="aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890AbCdEfGhIjKlMnOpQrStUvWxYz12"' >> ~/.zshrc

# Reload configuration
source ~/.zshrc

For temporary use (current session only):

# Just export them in your current terminal
export HCP_PROJECT_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"
export HCP_CLIENT_ID="abc123def456ghi789jkl012mno345pq"
export HCP_CLIENT_SECRET="aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890AbCdEfGhIjKlMnOpQrStUvWxYz12"

Step 5: Verify Authentication

# Check if environment variables are set correctly
echo "Project ID: $HCP_PROJECT_ID"
echo "Client ID: $HCP_CLIENT_ID"
echo "Client Secret: ${HCP_CLIENT_SECRET:0:10}..." # Shows only first 10 chars for security

# Verify all three are set and not empty
if [ -z "$HCP_PROJECT_ID" ] || [ -z "$HCP_CLIENT_ID" ] || [ -z "$HCP_CLIENT_SECRET" ]; then
    echo "❌ Error: One or more HCP environment variables are not set"
else
    echo "✅ All HCP environment variables are set"
fi

# Test with a simple scan
vault-radar scan folder -p . -o test-scan.json -f json

Common Authentication Errors

Error: HCP_PROJECT_ID environment variable must be set

You haven’t exported the variables in your current terminal session
If using a .env file, you forgot to run source .env
Solution: Run source .env or export the variables directly
Error: 404 Not Found or failed to get config options
Vault Radar is not enabled in your HCP project
The project ID may be incorrect

Go to HCP console → Your project
Look for “Vault Radar” in the left sidebar
Click “Enable” or “Get Started” if you see it
Complete any setup steps
If you don’t see Vault Radar, check your HCP tier/subscription
Verify your HCP_PROJECT_ID is correct (check the URL or project settings)
Error: oauth2: "unauthorized" "Authentication failed."

Your HCP_CLIENT_ID or HCP_CLIENT_SECRET is incorrect
The service principal may have been deleted or disabled
The credentials may have expired or been rotated

Verify your credentials in HCP console
Check that HCP_CLIENT_ID is a long alphanumeric string (not just your username or org name)
Generate a new key for your service principal if needed
Make sure you copied the entire Client Secret (they’re very long!)

Error: failed to get OAuth token

Network connectivity issues
HCP service may be temporarily unavailable
Firewall or proxy blocking access to HCP

Check your internet connection and try again

Debugging Steps:

# 1. Verify variables are loaded
env | grep HCP

# 2. Check the format of your credentials
echo "Project ID length: ${#HCP_PROJECT_ID}"  # Should be 36 (UUID format)
echo "Client ID length: ${#HCP_CLIENT_ID}"    # Should be 32+
echo "Client Secret length: ${#HCP_CLIENT_SECRET}"  # Should be 60+

# 3. If using .env file, make sure you sourced it
source .env

# 4. Try the scan again
vault-radar scan folder -p . -o test-scan.json -f json

Troubleshooting Authentication

Can’t find Service Principals in HCP?

Make sure you’re in a project (not just the organization view)
You need appropriate permissions — contact your HCP organization admin if you can’t create service principals

Lost your Client Secret?

The Client Secret is only displayed once when you create the service principal or generate a new key
You cannot view it again in the HCP console

Create a new service principal (recommended if you’re unsure)
Generate a new key for the existing service principal (invalidates the old secret)
Check if you saved it in a password manager or .env file

Using multiple HCP projects?

You can switch between projects by changing the HCP_PROJECT_ID variable
Consider creating separate service principals for each project

Scanning your machine using Vault-Radar

Now that all the requirements are set we can begin scanning!

Scan Current Directory

# Scan all files in current directory (requires output file)
vault-radar scan folder -p . -o results.json -f json

Scan Specific Files

# Scan .env file
vault-radar scan folder -p .env -o env-scan.json -f json

# Scan YAML files
vault-radar scan folder -p config.yaml -o config-scan.json -f json

Scanning Local Files

Basic scan

# Scan current directory recursively
vault-radar scan folder -p . -o results.json

# Scan specific directory
vault-radar scan folder -p /path/to/project -o scan-results.json

Scan with Output

# Scan and save results to JSON
vault-radar scan folder -p . -o results.json -f json

# Scan and save results to CSV
vault-radar scan folder -p . -o results.csv -f csv

# Scan and save results to SARIF format
vault-radar scan folder -p . -o results.sarif -f sarif

Scan .envFiles

# Scan specific .env file
vault-radar scan folder -p .env -o env-scan.json -f json

# Scan current directory (will include .env files)
vault-radar scan folder -p . -o scan-results.json -f json

Scan YAML Configuration Files

# Scan current directory (will include YAML files)
vault-radar scan folder -p . -o results.json -f json

# Scan specific config file
vault-radar scan folder -p config/application.yaml -o config-scan.json -f json

Configuration

Create a configuration YAML file (create .vault-radar.yaml in your project root)

# .vault-radar.yaml
scan:
  # Paths to scan
  paths:
    - .

  # File patterns to include
  include:
    - "*.env"
    - "*.yaml"
    - "*.yml"
    - "*.json"
    - "*.py"
    - "*.js"
    - "*.ts"

  # Directories to exclude
  exclude:
    - "node_modules"
    - "venv"
    - ".git"
    - "__pycache__"
    - "dist"
    - "build"

  # Output format (json, sarif, csv)
  output_format: json

  # Output file
  output_file: vault-radar-results.json

# Risk levels to report (critical, high, medium, low)
risk_levels:
  - critical
  - high
  - medium

# Custom patterns (optional)
custom_patterns:
  - name: "Custom API Key"
    pattern: "custom_api_[a-zA-Z0-9]{32}"
    risk: high

Using Configuration Values: while vault-radar doesn’t support a --config flag, you can use the configuration file as a reference and pass the values as command-line arguments:

# Example using values from .vault-radar.yaml
vault-radar scan folder \
  -p . \
  -o vault-radar-results.json \
  -f json

Output Formats

JSON Output request

vault-radar scan folder -p . -o results.json -f json

The JSON output file

{
  "findings": [
    {
      "type": "AWS Access Key",
      "file": ".env",
      "line": 5,
      "risk": "critical",
      "value": "AKIA****************",
      "description": "AWS Access Key ID detected"
    },
    {
      "type": "Generic API Key",
      "file": "config.yaml",
      "line": 12,
      "risk": "high",
      "value": "sk_live_***************",
      "description": "API key pattern detected"
    }
  ],
  "summary": {
    "total_files_scanned": 45,
    "total_findings": 2,
    "critical": 1,
    "high": 1,
    "medium": 0,
    "low": 0
  }
}

SARIF Output (for CI/CD)

vault-radar scan folder -p . -o results.sarif -f sarif

Default CSV Output

vault-radar scan folder -p . -o results.csv -f csv
# or simply (csv is default)
vault-radar scan folder -p . -o results.csv

Detected Secret Types

What does vault radar actually do scan?

Cloud Provider Keys

AWS Access Keys
Azure Storage Keys
Google Cloud API Keys
IBM Cloud API Keys

API Keys & Tokens

GitHub Personal Access Tokens
GitLab Tokens
Slack Tokens
Stripe API Keys
SendGrid API Keys

Database Credentials

PostgreSQL connection strings
MySQL passwords
MongoDB URIs
Redis passwords

Private Keys

RSA Private Keys
SSH Private Keys
PGP Private Keys

Generic Secrets

JWT Tokens
Bearer Tokens
API Keys (generic patterns)
Passwords in configuration files

Recommended — Common Use Cases

Scan Before Commit

#!/bin/bash
# pre-commit-scan.sh

# Ensure HCP credentials are set
if [ -z "$HCP_PROJECT_ID" ]; then
    echo "Error: HCP_PROJECT_ID not set"
    exit 1
fi

echo "Scanning for secrets..."
vault-radar scan folder -p . -o scan-results.json -f json

if [ $? -ne 0 ]; then
    echo "❌ Secrets detected! Please remove them before committing."
    cat scan-results.json
    exit 1
fi

echo "✅ No secrets detected."

CI/CD Integration

# .github/workflows/security-scan.yml
name: Security Scan

on: [push, pull_request]

jobs:
  scan-secrets:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Install Vault Radar
        run: |
          wget https://releases.hashicorp.com/vault-radar/0.10.0/vault-radar_0.10.0_linux_amd64.zip
          unzip vault-radar_0.10.0_linux_amd64.zip
          sudo mv vault-radar /usr/local/bin/

      - name: Scan for secrets
        env:
          HCP_PROJECT_ID: ${{ secrets.HCP_PROJECT_ID }}
          HCP_CLIENT_ID: ${{ secrets.HCP_CLIENT_ID }}
          HCP_CLIENT_SECRET: ${{ secrets.HCP_CLIENT_SECRET }}
        run: |
          vault-radar scan folder -p . -o results.sarif -f sarif

      - name: Upload results
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: results.sarif

Scan Multiple Directories

#!/bin/bash
# scan-all.sh

directories=("src" "config" "scripts" "tests")

for dir in "${directories[@]}"; do
    echo "Scanning $dir..."
    vault-radar scan folder -p "$dir" -o "scan-$dir.json"
done

Baseline Comparison

# First scan - establish baseline
vault-radar scan folder -p . -o baseline.json -f json

# Later scan - compare against baseline (only new secrets reported)
vault-radar scan folder -p . -o new-findings.json -f json -b baseline.json

Best Practices

Some tips and habits to have in mind.

Regular Scanning

# Add to crontab for daily scans
0 2 * * * cd /path/to/project && vault-radar scan folder -p . -o daily-scan.jsonLinks & resources

Exclude False Positives: Create .vault-radar-ignore

# Ignore test files with dummy secrets
tests/fixtures/test-secrets.env
tests/mock-data.yaml

# Ignore documentation examples
docs/examples/*.md

Integrate with Git Hooks

# .git/hooks/pre-commit
#!/bin/bash

vault-radar scan folder -p . -o pre-commit-scan.json -f json

if [ $? -ne 0 ]; then
    echo "Secrets detected!"
    cat pre-commit-scan.json
    exit 1
fi

Remediation Workflow 🔥 When secrets are found:

Rotate the exposed secret immediately
Remove from code and use environment variables
Update .gitignore to prevent future commits
Consider using HashiCorp Vault for secret management
Scan git history for the secret:

git log -p | grep -i "secret_pattern"

Example: Scanning This Project;

# Scan the current project for secrets
vault-radar scan folder -p /Users/alainairom/Devs/ibm-cos \
  -o ibm-cos-scan.json \
  -f json

# View results
cat ibm-cos-scan.json | jq '.findings'

# Count findings by risk level
cat ibm-cos-scan.json | jq '.summary'

Troubleshooting

Issue: Authentication Error

# Error: HCP_PROJECT_ID environment variable must be set

Solution;

# Set required environment variables
export HCP_PROJECT_ID="your-project-id"
export HCP_CLIENT_ID="your-client-id"
export HCP_CLIENT_SECRET="your-client-secret"

# Verify they are set
env | grep HCP

# Try scanning again
vault-radar scan folder -p . -o results.json -f json

Issue: Too Many Results

# Use baseline to track only new secrets
vault-radar scan folder -p . -o current.json -f json -b previous.json

# Limit number of secrets reported
vault-radar scan folder -p . -o results.json -f json -l 50

Issue: Slow Scanning

# Scan specific directories only
vault-radar scan folder -p src -o src-scan.json -f json
vault-radar scan folder -p config -o config-scan.json -f json

# Skip activeness checks for faster scanning
vault-radar scan folder -p . -o results.json -f json --skip-activeness

Issue: Need Quiet Output

# Disable UI output (summary not logged to stdout)
vault-radar scan folder -p . -o results.json -f json --disable-ui

Integration with Vault

After finding secrets with Vault Radar, migrate them to HashiCorp Vault:

# 1. Scan for secrets
vault-radar scan folder -p . -o findings.json

# 2. Extract secrets from findings
cat findings.json | jq -r '.findings[] | .file + ":" + (.line|tostring)'

# 3. Store in Vault (manual or scripted)
vault kv put secret/myapp/config \
  api_key="value-from-env-file" \
  db_password="value-from-config"

# 4. Update code to use Vault
# See VAULT_CONFIGURATION.md for integration details

Command Reference

# Basic scan (requires output file)
vault-radar scan folder -p <path> -o <output-file>

# With format (csv, json, or sarif)
vault-radar scan folder -p <path> -o <output-file> -f json

# With baseline comparison
vault-radar scan folder -p <path> -o <output-file> -b previous-scan.json

# With limit on number of secrets
vault-radar scan folder -p <path> -o <output-file> -l 100

# Disable UI output
vault-radar scan folder -p <path> -o <output-file> --disable-ui

# Skip activeness checks
vault-radar scan folder -p <path> -o <output-file> --skip-activeness

# Scan git repository
vault-radar scan repo -u <git-url>

# Version
vault-radar --version

# Help
vault-radar --help
vault-radar scan folder --help

Conclusion

Ultimately, securing your online applications begins long before the code reaches a production environment; it starts on your local machine. By integrating Vault Radar into your local development workflow, you transform a potentially “catastrophic” oversight into a proactive defense. Scanning locally allows you to catch hardcoded passwords, AWS keys, and PII in .env or YAML files before they are ever committed to a repository's history. This "shift-left" approach ensures that even if you confidently—but mistakenly—presume your code is clean, the radar identifies those hidden risks for you. Embracing this local-first security habit is the most effective way to prevent accidental leaks and ensure that your transition to the cloud is as secure as it is seamless.

>>> Thanks for reading <<<

Links and Resources

HashiCorp Cloud Platform: https://portal.cloud.hashicorp.com/sign-in
Vault Radar Documentation: https://developer.hashicorp.com/hcp/docs/vault-radar
Vault Radar CLI: https://developer.hashicorp.com/hcp/docs/vault-radar/cli
Vault Radar Demos: https://github.com/hashicorp-guides/vault-radar-demo
Vault Radar FAQ: https://developer.hashicorp.com/hcp/docs/vault-radar/faq
HashiCorp Radar Documentation: https://developer.hashicorp.com/vault/docs/radar
HashiCorp Security: https://www.hashicorp.com/en/trust/security
OWASP Cheat Sheet Series: https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html

The Bob Blueprint: Mastering IaC with Terraform and Ansible, Vault Security, and OTel Observability in One Click

Alain Airom (Ayrom) — Fri, 20 Mar 2026 16:26:43 +0000

Bulletproofing Bob: An End-to-End Deep Dive into a cloud-native application with Terraform, Ansible, Vault, and OpenTelemetry!

Introduction — Mastering the DevSecOps Lifecycle with “Bob”

The application showcases a complete end-to-end, cloud-native ecosystem where every stage of the application lifecycle — from infrastructure provisioning to real-time monitoring — is automated, secure, and observable, implemented and document by Bob, IBM’s SDLC.

Member Management Application

The core

The Member Management Application (affectionately dubbed “Bob”). Showcasing that in modern software, “it works on my machine” isn’t enough. We need a system that is:

Self-Healing & Scalable (Kubernetes/Docker).
Provisioned by Code (Terraform).
Configured Automatically (Ansible).
Fortified by Design (HashiCorp Vault).
Last but not least, observable by design (OpenTelemetry).
Press enter or click to view image in full size

from flask import Flask, render_template, request, jsonify, redirect, url_for
import sqlite3
import os
from datetime import datetime
import hashlib
import hvac
import re
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.sqlite3 import SQLite3Instrumentor

# Initialize OpenTelemetry
resource = Resource.create({"service.name": "member-management-app"})

# Trace provider
trace_provider = TracerProvider(resource=resource)
otlp_trace_exporter = OTLPSpanExporter(
    endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
    insecure=True
)
trace_provider.add_span_processor(BatchSpanProcessor(otlp_trace_exporter))
trace.set_tracer_provider(trace_provider)

# Metrics provider
metric_reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
        insecure=True
    )
)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)

# Initialize Flask app
app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv('FLASK_SECRET_KEY', 'dev-secret-key')

# Instrument Flask and SQLite
FlaskInstrumentor().instrument_app(app)
SQLite3Instrumentor().instrument()

# Get tracer and meter
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

# Create metrics
member_counter = meter.create_counter(
    "member.operations",
    description="Number of member operations",
    unit="1"
)

# Database configuration
DATABASE = os.getenv('DATABASE_PATH', 'members.db')

# Vault configuration
VAULT_ADDR = os.getenv('VAULT_ADDR', 'http://localhost:8200')
VAULT_TOKEN = os.getenv('VAULT_TOKEN', 'dev-token')
VAULT_MOUNT_POINT = os.getenv('VAULT_MOUNT_POINT', 'secret')

def get_vault_client():
    """Initialize and return Vault client"""
    try:
        client = hvac.Client(url=VAULT_ADDR, token=VAULT_TOKEN)
        if not client.is_authenticated():
            print("Warning: Vault authentication failed")
            return None
        return client
    except Exception as e:
        print(f"Warning: Could not connect to Vault: {e}")
        return None

def store_secret_in_vault(username, secret_type, secret_value):
    """Store a secret in Vault"""
    with tracer.start_as_current_span("store_secret_in_vault"):
        client = get_vault_client()
        if client:
            try:
                path = f"{VAULT_MOUNT_POINT}/data/members/{username}/{secret_type}"
                client.secrets.kv.v2.create_or_update_secret(
                    path=f"members/{username}/{secret_type}",
                    secret=dict(value=secret_value),
                    mount_point=VAULT_MOUNT_POINT
                )
                return True
            except Exception as e:
                print(f"Error storing secret in Vault: {e}")
                return False
        return False

def get_secret_from_vault(username, secret_type):
    """Retrieve a secret from Vault"""
    with tracer.start_as_current_span("get_secret_from_vault"):
        client = get_vault_client()
        if client:
            try:
                path = f"members/{username}/{secret_type}"
                secret = client.secrets.kv.v2.read_secret_version(
                    path=path,
                    mount_point=VAULT_MOUNT_POINT
                )
                return secret['data']['data']['value']
            except Exception as e:
                print(f"Error retrieving secret from Vault: {e}")
                return None
        return None

def hash_password(password):
    """Hash password using SHA-256"""
    return hashlib.sha256(password.encode()).hexdigest()

def get_db_connection():
    """Create database connection"""
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    """Initialize database with schema and sample data"""
    with tracer.start_as_current_span("init_db"):
        conn = get_db_connection()
        cursor = conn.cursor()

        # Create table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS members (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                last_name TEXT NOT NULL,
                first_name TEXT NOT NULL,
                date_of_birth TEXT NOT NULL,
                username TEXT UNIQUE NOT NULL,
                password_hash TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # Check if we need to add sample data
        cursor.execute('SELECT COUNT(*) FROM members')
        count = cursor.fetchone()[0]

        if count == 0:
            # Insert 5 sample records
            sample_members = [
                ('Smith', 'John', '1985-03-15', 'jsmith', 'password123', 'secret_data_1'),
                ('Johnson', 'Emily', '1990-07-22', 'ejohnson', 'password456', 'secret_data_2'),
                ('Williams', 'Michael', '1988-11-30', 'mwilliams', 'password789', 'secret_data_3'),
                ('Brown', 'Sarah', '1992-05-18', 'sbrown', 'passwordabc', 'secret_data_4'),
                ('Davis', 'Robert', '1987-09-25', 'rdavis', 'passwordxyz', 'secret_data_5')
            ]

            for last_name, first_name, dob, username, password, secret in sample_members:
                password_hash = hash_password(password)
                cursor.execute('''
                    INSERT INTO members (last_name, first_name, date_of_birth, username, password_hash)
                    VALUES (?, ?, ?, ?, ?)
                ''', (last_name, first_name, dob, username, password_hash))

                # Store secret in Vault
                store_secret_in_vault(username, 'secret', secret)

        conn.commit()
        conn.close()

@app.route('/')
def index():
    """Home page - list all members with optional search"""
    with tracer.start_as_current_span("index"):
        search_query = request.args.get('search', '').strip()
        conn = get_db_connection()

        if search_query:
            # Convert wildcard pattern to SQL LIKE pattern
            # Replace * with % for SQL LIKE
            sql_pattern = search_query.replace('*', '%')

            # Search in last_name, first_name, and username
            members = conn.execute('''
                SELECT * FROM members
                WHERE last_name LIKE ?
                   OR first_name LIKE ?
                   OR username LIKE ?
                ORDER BY last_name, first_name
            ''', (sql_pattern, sql_pattern, sql_pattern)).fetchall()

            member_counter.add(1, {"operation": "search"})
        else:
            members = conn.execute('SELECT * FROM members ORDER BY last_name, first_name').fetchall()
            member_counter.add(1, {"operation": "list"})

        conn.close()
        return render_template('index.html', members=members, search_query=search_query)

@app.route('/member/<int:member_id>')
def view_member(member_id):
    """View member details"""
    with tracer.start_as_current_span("view_member"):
        conn = get_db_connection()
        member = conn.execute('SELECT * FROM members WHERE id = ?', (member_id,)).fetchone()
        conn.close()

        if member is None:
            return "Member not found", 404

        # Retrieve secret from Vault
        secret = get_secret_from_vault(member['username'], 'secret')

        member_counter.add(1, {"operation": "view"})
        return render_template('view_member.html', member=member, secret=secret)

@app.route('/member/new', methods=['GET', 'POST'])
def create_member():
    """Create new member"""
    with tracer.start_as_current_span("create_member"):
        if request.method == 'POST':
            last_name = request.form['last_name']
            first_name = request.form['first_name']
            date_of_birth = request.form['date_of_birth']
            username = request.form['username']
            password = request.form['password']
            secret = request.form.get('secret', '')

            password_hash = hash_password(password)

            conn = get_db_connection()
            try:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO members (last_name, first_name, date_of_birth, username, password_hash)
                    VALUES (?, ?, ?, ?, ?)
                ''', (last_name, first_name, date_of_birth, username, password_hash))
                conn.commit()

                # Store secret in Vault
                if secret:
                    store_secret_in_vault(username, 'secret', secret)

                member_counter.add(1, {"operation": "create"})
                return redirect(url_for('index'))
            except sqlite3.IntegrityError:
                return "Username already exists", 400
            finally:
                conn.close()

        return render_template('create_member.html')

@app.route('/member/<int:member_id>/edit', methods=['GET', 'POST'])
def edit_member(member_id):
    """Edit member"""
    with tracer.start_as_current_span("edit_member"):
        conn = get_db_connection()
        member = conn.execute('SELECT * FROM members WHERE id = ?', (member_id,)).fetchone()

        if member is None:
            conn.close()
            return "Member not found", 404

        if request.method == 'POST':
            last_name = request.form['last_name']
            first_name = request.form['first_name']
            date_of_birth = request.form['date_of_birth']
            password = request.form.get('password')
            secret = request.form.get('secret')

            if password:
                password_hash = hash_password(password)
                conn.execute('''
                    UPDATE members 
                    SET last_name = ?, first_name = ?, date_of_birth = ?, password_hash = ?, updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (last_name, first_name, date_of_birth, password_hash, member_id))
            else:
                conn.execute('''
                    UPDATE members 
                    SET last_name = ?, first_name = ?, date_of_birth = ?, updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (last_name, first_name, date_of_birth, member_id))

            conn.commit()

            # Update secret in Vault if provided
            if secret:
                store_secret_in_vault(member['username'], 'secret', secret)

            conn.close()
            member_counter.add(1, {"operation": "update"})
            return redirect(url_for('view_member', member_id=member_id))

        # Retrieve secret from Vault
        secret = get_secret_from_vault(member['username'], 'secret')
        conn.close()

        return render_template('edit_member.html', member=member, secret=secret)

@app.route('/member/<int:member_id>/delete', methods=['POST'])
def delete_member(member_id):
    """Delete member"""
    with tracer.start_as_current_span("delete_member"):
        conn = get_db_connection()
        member = conn.execute('SELECT username FROM members WHERE id = ?', (member_id,)).fetchone()

        if member:
            conn.execute('DELETE FROM members WHERE id = ?', (member_id,))
            conn.commit()
            member_counter.add(1, {"operation": "delete"})

        conn.close()
        return redirect(url_for('index'))

@app.route('/admin')
def admin():
    """Admin dashboard with links to monitoring tools"""
    with tracer.start_as_current_span("admin"):
        return render_template('admin.html')

@app.route('/health')
def health():
    """Health check endpoint"""
    return jsonify({"status": "healthy", "timestamp": datetime.utcnow().isoformat()})

@app.route('/metrics')
def metrics_endpoint():
    """Metrics endpoint for Prometheus"""
    return "Metrics are exported via OTLP", 200

if __name__ == '__main__':
    init_db()
    port = int(os.getenv('PORT', 8080))
    app.run(host='0.0.0.0', port=port, debug=False)

# Made with Bob

Infrastructure as Code: The Foundation

Demonstrating a deployment starts with Terraform to provision the Kubernetes resources (Namespaces, PVCs, and Services).

Benefit: Consistency across environments — whether it’s Minikube for local dev or a cloud provider for production.

Terraform serves as the architect, defining the entire environment before a single line of application code runs.

Provider Management: It initializes the Kubernetes provider to interact with clusters like Minikube.
Resource Orchestration: Terraform handles the creation of Namespaces, ConfigMaps, and Secrets to ensure the environment is ready for deployment.
Deployment Automation: It defines the specifications for the application pods, including replica counts, resource limits (512Mi memory/500m CPU), and container images.
Service Exposure: It automatically provisions NodePort services for the App (30080), Prometheus (30090), and Grafana (30030).

terraform {
  required_version = ">= 1.0"

  required_providers {
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.23"
    }
    helm = {
      source  = "hashicorp/helm"
      version = "~> 2.11"
    }
  }
}

provider "kubernetes" {
  config_path    = "~/.kube/config"
  config_context = "minikube"
}

provider "helm" {
  kubernetes {
    config_path    = "~/.kube/config"
    config_context = "minikube"
  }
}

# Create namespace
resource "kubernetes_namespace" "member_management" {
  metadata {
    name = "member-management"
    labels = {
      name = "member-management"
      app  = "member-management"
    }
  }
}

# Create ConfigMaps
resource "kubernetes_config_map" "app_config" {
  metadata {
    name      = "app-config"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  data = {
    DATABASE_PATH                = "/data/members.db"
    PORT                         = "8080"
    VAULT_ADDR                   = "http://vault:8200"
    VAULT_MOUNT_POINT            = "secret"
    OTEL_EXPORTER_OTLP_ENDPOINT  = "http://otel-collector:4317"
  }
}

resource "kubernetes_config_map" "otel_collector_config" {
  metadata {
    name      = "otel-collector-config"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  data = {
    "otel-collector-config.yaml" = file("${path.module}/../otel-collector-config.yaml")
  }
}

resource "kubernetes_config_map" "prometheus_config" {
  metadata {
    name      = "prometheus-config"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  data = {
    "prometheus.yml" = file("${path.module}/../prometheus.yml")
  }
}

resource "kubernetes_config_map" "grafana_datasources" {
  metadata {
    name      = "grafana-datasources"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  data = {
    "datasources.yml" = file("${path.module}/../grafana-datasources.yml")
  }
}

# Create Secrets
resource "kubernetes_secret" "app_secrets" {
  metadata {
    name      = "app-secrets"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  data = {
    VAULT_TOKEN       = base64encode("dev-token")
    FLASK_SECRET_KEY  = base64encode("production-secret-key-change-me")
  }

  type = "Opaque"
}

# Create PersistentVolumeClaims
resource "kubernetes_persistent_volume_claim" "app_data" {
  metadata {
    name      = "app-data-pvc"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  spec {
    access_modes = ["ReadWriteOnce"]
    resources {
      requests = {
        storage = "1Gi"
      }
    }
    storage_class_name = "standard"
  }
}

resource "kubernetes_persistent_volume_claim" "prometheus_data" {
  metadata {
    name      = "prometheus-data-pvc"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  spec {
    access_modes = ["ReadWriteOnce"]
    resources {
      requests = {
        storage = "5Gi"
      }
    }
    storage_class_name = "standard"
  }
}

resource "kubernetes_persistent_volume_claim" "grafana_data" {
  metadata {
    name      = "grafana-data-pvc"
    namespace = kubernetes_namespace.member_management.metadata[0].name
  }

  spec {
    access_modes = ["ReadWriteOnce"]
    resources {
      requests = {
        storage = "2Gi"
      }
    }
    storage_class_name = "standard"
  }
}

Security First: Locking Down the Vault

Security isn’t an afterthought; it’s baked into the core architecture via HashiCorp Vault. Detail the Security Architecture:

Password Hashing: Using SHA−256 for member passwords.
Secret Management: Integration with HashiCorp Vault (Port 8200) to store and retrieve sensitive member data.
K8s Secrets: Utilizing Base64 encoding and encryption for cluster-level security.
Centralized Secrets: Instead of hardcoding credentials, the application retrieves sensitive data from Vault’s KV v2 engine.
Identity & Access: The system uses Token Authentication (dev-token) to verify the application's identity before granting access to secrets.
Encryption at Rest: Vault ensures that sensitive member data is protected, while the application itself hashes passwords using SHA-256 for a multi-layered defense.
Integration: The Flask app communicates with the Vault API on port 8200 to store and retrieve member-specific secrets during the creation and retrieval flows.

# Vault configuration
VAULT_ADDR = os.getenv('VAULT_ADDR', 'http://localhost:8200')
VAULT_TOKEN = os.getenv('VAULT_TOKEN', 'dev-token')
VAULT_MOUNT_POINT = os.getenv('VAULT_MOUNT_POINT', 'secret')

def get_vault_client():
    """Initialize and return Vault client"""
    try:
        client = hvac.Client(url=VAULT_ADDR, token=VAULT_TOKEN)
        if not client.is_authenticated():
            print("Warning: Vault authentication failed")
            return None
        return client
    except Exception as e:
        print(f"Warning: Could not connect to Vault: {e}")
        return None

def store_secret_in_vault(username, secret_type, secret_value):
    """Store a secret in Vault"""
    with tracer.start_as_current_span("store_secret_in_vault"):
        client = get_vault_client()
        if client:
            try:
                path = f"{VAULT_MOUNT_POINT}/data/members/{username}/{secret_type}"
                client.secrets.kv.v2.create_or_update_secret(
                    path=f"members/{username}/{secret_type}",
                    secret=dict(value=secret_value),
                    mount_point=VAULT_MOUNT_POINT
                )
                return True
            except Exception as e:
                print(f"Error storing secret in Vault: {e}")
                return False
        return False

def get_secret_from_vault(username, secret_type):
    """Retrieve a secret from Vault"""
    with tracer.start_as_current_span("get_secret_from_vault"):
        client = get_vault_client()
        if client:
            try:
                path = f"members/{username}/{secret_type}"
                secret = client.secrets.kv.v2.read_secret_version(
                    path=path,
                    mount_point=VAULT_MOUNT_POINT
                )
                return secret['data']['data']['value']
            except Exception as e:
                print(f"Error retrieving secret from Vault: {e}")
                return None
        return None

def hash_password(password):
    """Hash password using SHA-256"""
    return hashlib.sha256(password.encode()).hexdigest()

Post-Deployment Magic with Ansible

Lifecycle management is essential. While Terraform builds the “house,” Ansible handles the “furniture.”

Inventory Management: Ansible maintains a structured inventory of the target hosts, grouping them into application and monitoring tiers.

all:
  hosts:
    localhost:
      ansible_connection: local
      ansible_python_interpreter: /usr/bin/python3

  vars:
    project_root: "{{ playbook_dir }}/.."
    namespace: member-management
    docker_image: member-management-app
    docker_tag: latest

# Made with Bob

Automated Deployment: The deploy.yml playbook automates the pull of Docker images and the startup of the entire container stack.

---
- name: Deploy Member Management Application
  hosts: localhost
  gather_facts: yes

  tasks:
    - name: Check if Minikube is running
      command: minikube status
      register: minikube_status
      ignore_errors: yes
      changed_when: false

    - name: Start Minikube if not running
      command: minikube start
      when: minikube_status.rc != 0

    - name: Build Docker image
      command: docker build -t {{ docker_image }}:{{ docker_tag }} {{ project_root }}
      args:
        chdir: "{{ project_root }}"

    - name: Load Docker image into Minikube
      command: minikube image load {{ docker_image }}:{{ docker_tag }}

    - name: Create namespace
      kubernetes.core.k8s:
        state: present
        definition:
          apiVersion: v1
          kind: Namespace
          metadata:
            name: "{{ namespace }}"
            labels:
              name: "{{ namespace }}"
              app: "{{ namespace }}"

    - name: Apply Kubernetes manifests
      kubernetes.core.k8s:
        state: present
        src: "{{ item }}"
        namespace: "{{ namespace }}"
      loop:
        - "{{ project_root }}/k8s/configmap.yaml"
        - "{{ project_root }}/k8s/secrets.yaml"
        - "{{ project_root }}/k8s/pvc.yaml"
        - "{{ project_root }}/k8s/vault-deployment.yaml"
        - "{{ project_root }}/k8s/otel-collector-deployment.yaml"
        - "{{ project_root }}/k8s/prometheus-deployment.yaml"
        - "{{ project_root }}/k8s/grafana-deployment.yaml"
        - "{{ project_root }}/k8s/app-deployment.yaml"

    - name: Wait for Vault to be ready
      kubernetes.core.k8s_info:
        kind: Pod
        namespace: "{{ namespace }}"
        label_selectors:
          - app=vault
        wait: yes
        wait_condition:
          type: Ready
          status: "True"
        wait_timeout: 300

    - name: Wait for application to be ready
      kubernetes.core.k8s_info:
        kind: Pod
        namespace: "{{ namespace }}"
        label_selectors:
          - app=member-management-app
        wait: yes
        wait_condition:
          type: Ready
          status: "True"
        wait_timeout: 300

    - name: Get Minikube IP
      command: minikube ip
      register: minikube_ip
      changed_when: false

    - name: Display access URLs
      debug:
        msg:
          - "Application URL: http://{{ minikube_ip.stdout }}:30080"
          - "Prometheus URL: http://{{ minikube_ip.stdout }}:30090"
          - "Grafana URL: http://{{ minikube_ip.stdout }}:30030"
          - "Grafana credentials: admin/admin"

# Made with Bob

Seamless Updates: The update.yml playbook allows for rolling updates, ensuring that the application can be patched or upgraded with minimal downtime.
Orchestration: It ensures dependencies — like ensuring Vault is unsealed and OTel is running — are handled in the correct sequence.

Observability: Seeing Inside the Black Box

This is where Bob shines; the Observability Stack:

OpenTelemetry (OTel): The application sends traces and metrics to an OTel Collector.

OTel Collector: A central collector (Port 4317/4318) receives telemetry data, acting as a vendor-neutral gateway.

Prometheus: the collector exports data to Prometheus, where it is stored as time-series data for analysis.
Grafana: serves as the “single pane of glass,” providing pre-configured dashboards that track member operations, system health, and request latencies.
Instrumentation: the Flask application and SQLite database are instrumented to produce traces and metrics automatically.

# Trace provider
trace_provider = TracerProvider(resource=resource)
otlp_trace_exporter = OTLPSpanExporter(
    endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
    insecure=True
)
trace_provider.add_span_processor(BatchSpanProcessor(otlp_trace_exporter))
trace.set_tracer_provider(trace_provider)

# Metrics provider
metric_reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
        insecure=True
    )
)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)

Conclusion

In summary, “Bob” represents a comprehensive shift from manual deployment to a fully automated, production-ready ecosystem. Starting from the ground up, the application provides a functional and intuitive UI for complete database record management, allowing users to handle member data with ease. This application layer is seamlessly integrated with HashiCorp Vault, ensuring that every piece of sensitive information is handled with industry-standard security through centralized secret management and robust encryption. By combining a reliable Python-based backend with a modern frontend, Bob proves that internal business tools can be both powerful and user-friendly right out of the box.

The true power of Bob, however, lies in its sophisticated infrastructure and lifecycle automation. Through the use of Terraform, the entire Kubernetes environment — including networking, storage, and pod configurations — is defined as code, ensuring consistent deployments across any cluster. Ansible further enhances this by orchestrating the operational lifecycle, from initial image builds to rolling updates. This is all capped off by an “out-of-the-box” observability stack where OpenTelemetry, Prometheus, and Grafana work in unison to provide real-time insights into the system’s health. Ultimately, Bob demonstrates that with the right integration of IaC, security, and monitoring, a complex cloud-native application can be deployed and managed with a single command.

>>> Thanks for reading <<<

DEV Community: Alain Airom (Ayrom)

Protecting the Crown Jewels: Securing Source Code in Containerized Applications

Introduction and some background

What are the Technics that Bob Provided?

Why This Matters?

Choosing Your Shield: Language-Specific Protection Strategies

The Spectrum of Security

Key Takeaways for Implementation

Go Sample Application (one of the good choices)

Python Sample Application (moderate choice)

Links

The “The Architecture Handbook for Milvus Vector Database” Book Review

Introduction-Mastering the Core of AI Memory

A synthesis of the Book

Setup and Configuration

Data Model — Reading Data, Modification and Compacting Data

Scalabilty and Performance

Advanced Topics

Conclusion

Links

Stop Treating Your Code Security Like a “Check Engine” Light (or part #2 of what we can do with Bob even if we’re not coders…)

Introduction

Lessons Learned and Conclusion

Links

My first Experience with Google’s Agent Development Kit (ADK-Go)

Introduction

Excerpt from ADK-Go Repository

Running the Provided Sample Out-of-the-Box!

Implementing A local Agent using Ollama and Lessons Learned

First Test: Console Mode Application

Lessons learned while making this application

Second Test: Web UI Mode Application

Conclusion

Links

Things you can do with IBM Bob if you don’t want to Code!

Introduction

TL;DR-What is Instana?

When Bob meets Stan 😂

Conclusion

Links

Granite 🪨 4.0 3B 👓 Vision: Compact Multimodal Intelligence for Enterprise Documents is out and Bob built a UI to use it!

Introduction

Supported Tasks

Implementation

Using the code provided on Hugging Face

Streamlit version including Docling (no Bob)

Bob’s version built on Gradio including Docling

Conclusion

Links

Testing ‘chunkweaver’

Introduction and Disclaimer

My local Implementation

The main application

Milvus Vector DB Implementation

The chat implementation using Ollama

Documents I used to enrich the RAG

Conclusion

Links

Building Smarter Agents: A Guide to IBM Bob’s “MCP Builder with Agent Utils”

Building Smarter Agents: A Guide to IBM’s “MCP Builder with Agent Utils”

The Foundation: BaseMCP and the Tool Registry

Anatomy of an MCP Tool

From Development to Production

Best Practices for Scalable Agents

Conclusion

Links

Discovering “GraphQLite” from Colliery-IO 📈

Intrduction and Excerpt from GraphQLite Repository

Moving Beyond Simple Search: Building with GraphQLite

The Application

Processing and Back-End Engine

The Command Center: “index.html"

Conclusion

Links

Building an Agent with MCP for Docling XBRL Document Conversion (by Bob)

TL;DR-What is XBRL?

The Docling XBRL Processing Pipeline

Configuring the Offline Backend

The Conversion Lifecycle

How parsing (for example with an agent) to read XBRL documents can help?

The Command Center: `“index.html"`