DEV Community: MUHAMMAD MUSADIQ

How to Build a Hybrid Workflow Engine: Combining Finite State Machines with LLM Reasoning Nodes

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 04:17:00 +0000

Building AI applications purely on LLM chains usually ends in tears. One day, the model outputs crisp JSON, and the next, it returns a polite paragraph that breaks your downstream code. On the flip side, traditional state machines are bulletproof but rigid—they fail the moment a user submits ambiguous text or an unpredictable input.

I spent weeks debugging broken multi-prompt pipelines before realizing the solution was simple: we need to fuse deterministic state machines with LLM reasoning nodes. By using a durable workflow engine like Temporal to manage state transitions while letting LLMs make bounded decisions, you get reliable, deterministic AI systems that won't crash in production.

Here is how you can build a hybrid workflow engine from scratch.

A state machine needs explicit states, valid transitions, and strict rules. An LLM node takes raw context, evaluates it, and returns a single decision from a pre-defined set of valid actions.

In a hybrid setup, the state machine handles orchestration, state persistence, retries, and execution limits. The LLM acts purely as a routing function inside a single state. The LLM never updates global state directly. It simply suggests the next transition based on structured data.

If you are scaling enterprise systems, setting up robust workflow automation early prevents massive tech debt later.

+-----------------------------------------------------------+
|                    Temporal Workflow                      |
|                                                           |
|  [State: DRAFT] ---> [Activity: LLM Evaluation Node]      |
|                                |                          |
|                                v                          |
|                      Structured Decision                  |
|                        /              \                   |
|                       v                v                  |
|          [State: REQUIRES_HUMAN]   [State: APPROVED]      |
+-----------------------------------------------------------+

Step 2: Set Up Temporal and Dependencies

We will use Temporal for durable execution and Python for our workflow logic. First, spin up a local Temporal server using the Temporal CLI or Docker.

Install the required Python packages:

pip install temporalio openai pydantic

Make sure your Temporal server is running locally on port 7233.

Step 3: Define Rigid State Models with Pydantic

To enforce deterministic AI behavior, the LLM must return strict schema outputs. Never let the LLM output freeform text to control workflow state.

Create a file named schemas.py:

from enum import Enum
from pydantic import BaseModel, Field

class WorkflowState(str, Enum):
    DRAFT = "DRAFT"
    NEEDS_REVISION = "NEEDS_REVISION"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"

class LLMDecision(str, Enum):
    APPROVE = "APPROVE"
    REQUEST_CHANGES = "REQUEST_CHANGES"
    REJECT = "REJECT"

class EvaluationResult(BaseModel):
    decision: LLMDecision
    reasoning: str = Field(description="Brief reason for the decision")
    confidence_score: float = Field(description="Confidence between 0.0 and 1.0")

Step 4: Build the LLM Reasoning Activity

Temporal activities handle side effects like API calls or database operations. If an LLM call fails or hits a rate limit, Temporal automatically handles retries without restarting the whole workflow.

When teams start building complex AI agents, keeping side effects inside isolated activities is the single best decision they can make.

Create activities.py:

import json
import os
from openai import OpenAI
from temporalio import activity
from schemas import EvaluationResult, LLMDecision

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@activity.defn
async def evaluate_document_activity(document_text: str) -> EvaluationResult:
    prompt = f"""
    Analyze the following text for compliance and quality.
    Select APPROVE if it meets standard criteria.
    Select REQUEST_CHANGES if minor edits are needed.
    Select REJECT if it violates basic guidelines.

    Document Text:
    {document_text}
    """

    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a deterministic logic node in a workflow engine."},
            {"role": "user", "content": prompt}
        ],
        response_format=EvaluationResult
    )

    return response.choices[0].message.parsed

Step 5: Implement the State Machine Workflow

Now write the workflow that processes state transitions based on the output of the LLM node. Temporal guarantees that workflow state persists across crashes, server restarts, or network failures.

Create workflows.py:

from datetime import timedelta
from temporalio import workflow
from schemas import WorkflowState, LLMDecision, EvaluationResult

with workflow.unsafe.imports_passed_through():
    from activities import evaluate_document_activity

@workflow.defn
class DocumentApprovalWorkflow:
    def __init__(self):
        self.current_state = WorkflowState.DRAFT
        self.history = []

    @workflow.run
    async def run(self, document_text: str) -> dict:
        self.history.append(f"Started in state: {self.current_state}")

        # Execute LLM Reasoning Node
        result: EvaluationResult = await workflow.execute_activity(
            evaluate_document_activity,
            document_text,
            start_to_close_timeout=timedelta(seconds=30)
        )

        # Enforce State Machine Transitions based on LLM output
        if result.decision == LLMDecision.APPROVE and result.confidence_score > 0.8:
            self.current_state = WorkflowState.APPROVED
        elif result.decision == LLMDecision.REQUEST_CHANGES:
            self.current_state = WorkflowState.NEEDS_REVISION
        else:
            # Low confidence or explicit reject routes to REJECTED
            self.current_state = WorkflowState.REJECTED

        self.history.append(f"Transitioned to state: {self.current_state} | Reason: {result.reasoning}")

        return {
            "final_state": self.current_state,
            "confidence": result.confidence_score,
            "history": self.history
        }

Step 6: Create the Worker and Trigger the Workflow

To run the workflow, set up a worker process to listen on a Temporal task queue.

Create run_worker.py:

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from activities import evaluate_document_activity
from workflows import DocumentApprovalWorkflow

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="hybrid-workflow-queue",
        workflows=[DocumentApprovalWorkflow],
        activities=[evaluate_document_activity],
    )
    print("Worker started. Listening for tasks...")
    await worker.run()

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

In a separate terminal, trigger the execution using execute_workflow.py:

import asyncio
from temporalio.client import Client
from workflows import DocumentApprovalWorkflow

async def main():
    client = await Client.connect("localhost:7233")

    sample_doc = "The policy document outlines terms for user data retention. All data is deleted after 30 days."

    result = await client.execute_workflow(
        DocumentApprovalWorkflow.run,
        sample_doc,
        id="doc-approval-001",
        task_queue="hybrid-workflow-queue",
    )

    print("Workflow Execution Result:")
    print(result)

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

Run the worker terminal first, then run the execution script. You will see Temporal step through the workflow states deterministically, using the LLM output strictly as a routing signal.

Moving Beyond Simple Logic

Building hybrid engines keeps your production code sane. State machines provide observability, deterministic retries, and guardrails. LLMs supply flexible evaluation and contextual understanding. Mixing them prevents unpredictable loop failures and runaway API bills.

If you are expanding your infrastructure and need expert developers to scale these systems, Gaper connects companies with vetted software engineers who specialize in production ready workflow orchestration and AI implementations.

How to Implement Human-in-the-Loop Approval Gates for Risky AI Agent Tool Calls

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 05:03:10 +0000

Autonomous AI agents are great until they drop a database table or send an unintended email to thousands of clients. Letting an LLM execute arbitrary actions without human validation is a fast track to production incidents. To keep your systems safe, you need an approval gate that pauses high-risk tool calls and waits for explicit human confirmation before executing sensitive operations.

Python 3.9 or higher installed on your machine.
Basic familiarity with Python functions and dictionaries.
No external libraries required, standard Python libraries like functools and typing work fine.

Step 1: Flag Risky Tools with Metadata

Start by defining custom tool decorators or metadata to tag functions as high risk. You do not need complex frameworks to do this. A simple decorator that attaches a requires_approval attribute to sensitive functions keeps your code clean.

import functools

def ai_tool(requires_approval: bool = False):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        wrapper.requires_approval = requires_approval
        wrapper.is_ai_tool = True
        return wrapper
    return decorator

# Example tools
@ai_tool(requires_approval=False)
def search_knowledge_base(query: str) -> str:
    return f"Results for: {query}"

@ai_tool(requires_approval=True)
def delete_user_account(user_id: str) -> str:
    return f"User {user_id} successfully deleted."

Step 2: Build the Interceptor Routine

Next, write an execution layer that inspects the target tool before running it. If the function has requires_approval=True, the wrapper suspends execution and requests user consent via console input, API callback, or Slack notification.

Here is a lightweight CLI implementation:

def execute_tool(tool_func, *args, **kwargs):
    tool_name = tool_func.__name__
    needs_review = getattr(tool_func, "requires_approval", False)

    if needs_review:
        print(f"\n[APPROVAL REQUIRED] Agent requested to call: {tool_name}")
        print(f"Arguments: args={args}, kwargs={kwargs}")

        user_input = input("Approve execution? (y/N): ").strip().lower()
        if user_input != 'y':
            print(f"[BLOCKED] Execution of {tool_name} was aborted by user.")
            return {"status": "rejected", "reason": "User denied approval"}

    print(f"[EXECUTING] Running {tool_name}...")
    result = tool_func(*args, **kwargs)
    return {"status": "success", "result": result}

This pattern works well whether you are building custom AI agent development pipelines or setting up complex enterprise rules. For instance, in sensitive fields like AI for Accounting & Finance, automated fund transfers or invoice modifications should always hit an explicit guardrail before moving real money.

Step 3: Handle Agent Responses and Execution

Now wire the tool executor directly into your agent call loop. When the model selects a tool, route the call through your execute_tool router rather than calling the tool directly.

# Simulated model tool calls coming from an agent framework
tool_calls_queue = [
    {"func": search_knowledge_base, "args": ("password reset steps",)},
    {"func": delete_user_account, "kwargs": {"user_id": "usr_9981"}}
]

for call in tool_calls_queue:
    func = call["func"]
    args = call.get("args", ())
    kwargs = call.get("kwargs", {})

    response = execute_tool(func, *args, **kwargs)
    print("Execution output:", response)

Integrating human approval gates like this forms the backbone of reliable workflow automation, giving teams confidence to deploy autonomous systems without giving up control over dangerous actions.

Expected Results

When running the full script, standard read operations execute immediately, while high-risk functions prompt for explicit approval:

[EXECUTING] Running search_knowledge_base...
Execution output: {'status': 'success', 'result': 'Results for: password reset steps'}

[APPROVAL REQUIRED] Agent requested to call: delete_user_account
Arguments: args=(), kwargs={'user_id': 'usr_9981'}
Approve execution? (y/N): n
[BLOCKED] Execution of delete_user_account was aborted by user.
Execution output: {'status': 'rejected', 'reason': 'User denied approval'}

If you answer y, the function executes and returns the success output. If you enter n or anything else, the process halts safely and returns a clear refusal message to the agent so it can handle the failure gracefully.

If you need help scaling agent architectures or embedding custom governance controls into your production stack, the team at Gaper can connect you with experienced engineers who build secure AI software.

How to Build a Multi-Tenant Data Isolation Layer for Enterprise AI Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 05:00:48 +0000

Deploying enterprise AI agents across multiple clients quickly becomes dangerous if your retrieval pipeline lacks strict tenant boundaries. Standard vector searches only evaluate semantic similarity, so without hard filtering, your agent can easily pull data from Tenant A to answer a prompt from Tenant B. I learned this the hard way when a test agent leaked a competitor's internal notes into a chat response during an early staging build. To fix this, you need a dedicated isolation layer that intercepts every agent tool call, injects validated tenant IDs, and enforces strict data boundaries at the database level.

Before writing code, make sure you have the following ready:

Python 3.10 or higher installed
A vector database supporting payload metadata filtering, such as Qdrant or PostgreSQL with pgvector
FastAPI or Flask for your web backend
An identity provider issuing JWTs that contain tenant metadata

Step 1: Create a Thread-Safe Tenant Context

Your system needs a reliable way to track who is making the request across asynchronous function calls. Do not rely on the LLM to pass tenant IDs in tool parameters. The model can hallucinate parameters or get tricked via prompt injection.

Use Python's contextvars module to store the active tenant ID securely in memory for each request context.

import contextvars

# Context variable scoped to the current execution thread or async task
_tenant_context = contextvars.ContextVar("tenant_id", default=None)

def set_tenant(tenant_id: str):
    return _tenant_context.set(tenant_id)

def get_tenant() -> str:
    tenant_id = _tenant_context.get()
    if not tenant_id:
        raise PermissionError("Access denied. No active tenant context found.")
    return tenant_id

Step 2: Build Middleware to Extract and Set Context

Intercept incoming API calls before they ever reach your agent runtime. Read the user's JWT, extract the verified tenant identifier, and store it in your context variable.

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

def verify_and_parse_jwt(token: str) -> dict:
    # Decode and validate your JWT here using your auth provider's public key
    # Returning mock data for demonstration
    return {"tenant_id": "tenant_company_abc"}

@app.middleware("http")
async def tenant_isolation_middleware(request: Request, call_next):
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing authorization header.")

    raw_token = auth_header.split(" ")[1]
    claims = verify_and_parse_jwt(raw_token)

    # Store tenant context for the duration of this request lifecycle
    token_ref = set_tenant(claims["tenant_id"])
    try:
        response = await call_next(request)
        return response
    finally:
        # Clear context after request finishes
        _tenant_context.reset(token_ref)

Step 3: Inject System Filters into Vector Store Queries

When your agent triggers a retrieval tool, your vector database wrapper must automatically attach metadata filters. If you are building custom AI workflows, teams often use custom tooling layers. You can explore specialized patterns at AI Agent Development to see how agents scale in production.

Here is how you inject the tenant filter programmatically:

def search_knowledge_base(query_vector: list, top_k: int = 5):
    # Fetch tenant ID directly from global request context
    active_tenant = get_tenant()

    # Enforce metadata filter regardless of what the LLM requested
    tenant_filter = {
        "must": [
            {"key": "tenant_id", "match": {"value": active_tenant}}
        ]
    }

    # Example Qdrant client query
    results = vector_client.search(
        collection_name="enterprise_knowledge",
        query_vector=query_vector,
        query_filter=tenant_filter,
        limit=top_k
    )
    return results

Step 4: Wrap Custom Agent Tools with Security Decorators

Agents usually do more than vector lookups. They execute SQL queries, send webhooks, or interact with external services. Create a decorator that wraps all agent tools and forces tenant check verification.

If your setup spans multiple legacy systems, reviewing AI implementation and consulting strategies can help map out cross-system data boundaries.

from functools import wraps

def enforce_tenant_boundary(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        current_tenant = get_tenant()
        print(f"[AUDIT LOG] Executing {func.__name__} for Tenant: {current_tenant}")

        # Inject verified tenant ID into tool parameters explicitly
        kwargs["tenant_id"] = current_tenant
        return func(*args, **kwargs)
    return wrapper

@enforce_tenant_boundary
def fetch_customer_invoices(customer_id: str, tenant_id: str = None):
    # The database query MUST use the injected tenant_id parameter
    query = "SELECT * FROM invoices WHERE customer_id = %s AND tenant_id = %s"
    return db.execute(query, (customer_id, tenant_id))

Expected Outcomes

After deploying this isolation layer, your system achieves several clear security guarantees:

Data leakage between tenants becomes impossible at the tool layer because filters are appended outside the LLM's control.
Prompt injection attacks trying to override tenant settings will fail completely.
Audit logs automatically capture every tool execution alongside a verified tenant identifier.

If you need extra engineering horsepower to design or implement secure AI architectures, Gaper connects companies with vetted developers ready to build enterprise software.

Building Model-Agnostic AI Agents: How to Implement Runtime Fallbacks Between OpenAI, Claude, and Gemini

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:55:11 +0000

Relying on a single LLM provider for a production application is risky. API rate limits, temporary outages, and unexpected latency spikes can bring down your application without warning. When your primary provider goes offline, your users shouldn't be stuck staring at a loading spinner. You need a system that automatically routes requests to backup providers like Claude or Gemini, without breaking your agent's state or changing your output format.

Here is how to build a lightweight, model-agnostic fallback handler in Python to keep your agents running smoothly.

Before starting, make sure you have:

Python 3.10 or higher installed.
API keys for OpenAI, Anthropic, and Google Gemini.
The provider SDKs installed (openai, anthropic, google-genai).

If you are building complex systems, setting up robust AI agent development patterns early saves you dozens of hours in maintenance down the road.

Step 1: Normalize Prompts and Outputs

Different providers expect different payload formats. OpenAI uses a messages list with role and content. Anthropic uses a similar system but handles system prompts as a top-level parameter. Gemini uses its own content structure.

Create a single standard format for your application.

from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ChatMessage:
    role: str # "system", "user", or "assistant"
    content: str

@dataclass
class LLMResponse:
    content: str
    provider: str
    model: str

Step 2: Build Generic Provider Wrappers

Wrap each client in a standard interface. Each function should accept your ChatMessage list and return your LLMResponse.

import os
from openai import OpenAI
from anthropic import Anthropic
import google.genai as genai

class ProviderAdapter:
    def __init__(self):
        self.openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.anthropic_client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))

    def call_openai(self, messages: List[ChatMessage], model="gpt-4o") -> LLMResponse:
        formatted = [{"role": m.role, "content": m.content} for m in messages]
        res = self.openai_client.chat.completions.create(model=model, messages=formatted)
        return LLMResponse(
            content=res.choices[0].message.content,
            provider="openai",
            model=model
        )

    def call_claude(self, messages: List[ChatMessage], model="claude-3-5-sonnet-20241022") -> LLMResponse:
        system_prompt = ""
        user_messages = []

        for m in messages:
            if m.role == "system":
                system_prompt += m.content + "\n"
            else:
                user_messages.append({"role": m.role, "content": m.content})

        res = self.anthropic_client.messages.create(
            model=model,
            max_tokens=1024,
            system=system_prompt.strip(),
            messages=user_messages
        )
        return LLMResponse(
            content=res.content[0].text,
            provider="anthropic",
            model=model
        )

    def call_gemini(self, messages: List[ChatMessage], model="gemini-2.5-flash") -> LLMResponse:
        # Flatten system and user messages into a simple text prompt for basic compatibility
        prompt_text = "\n".join([f"{m.role}: {m.content}" for m in messages])
        res = self.gemini_client.models.generate_content(
            model=model,
            contents=prompt_text
        )
        return LLMResponse(
            content=res.text,
            provider="gemini",
            model=model
        )

Step 3: Write the Fallback Router Loop

Now build the fallback router. It loops through your preferred order of models. If one fails due to an API error, timeout, or rate limit, it catches the exception, logs a warning, and moves to the next model immediately.

import logging

logging.basicConfig(level=logging.INFO)

class ResilientAgent:
    def __init__(self):
        self.adapter = ProviderAdapter()
        # Define fallback hierarchy: (provider_name, model_name)
        self.fallback_chain = [
            ("openai", "gpt-4o"),
            ("anthropic", "claude-3-5-sonnet-20241022"),
            ("gemini", "gemini-2.5-flash")
        ]

    def execute(self, messages: List[ChatMessage]) -> LLMResponse:
        for provider, model in self.fallback_chain:
            try:
                logging.info(f"Attempting execution with {provider} ({model})...")

                if provider == "openai":
                    return self.adapter.call_openai(messages, model=model)
                elif provider == "anthropic":
                    return self.adapter.call_claude(messages, model=model)
                elif provider == "gemini":
                    return self.adapter.call_gemini(messages, model=model)

            except Exception as e:
                logging.warning(f"Provider {provider} failed with error: {e}. Trying fallback...")

        raise RuntimeError("All LLM providers failed. Check API status and keys.")

This pattern is useful when building mission critical system integration and workflow automation, where uptime is mandatory for daily business operations.

Step 4: Test Your Fallback Flow

You can run a quick test script to make sure the fallback triggers when the primary provider encounters an issue.

if __name__ == "__main__":
    agent = ResilientAgent()

    conversation = [
        ChatMessage(role="system", content="You are a helpful assistant."),
        ChatMessage(role="user", content="Explain runtime fallbacks in one sentence.")
    ]

    # Simulating a failure by supplying a bogus key for OpenAI in env variables
    # will force the agent to skip OpenAI and hit Claude or Gemini automatically.

    try:
        response = agent.execute(conversation)
        print("\n--- Success ---")
        print(f"Provider Used: {response.provider}")
        print(f"Model Used: {response.model}")
        print(f"Response: {response.content}")
    except Exception as err:
        print(f"Execution failed completely: {err}")

Expected Outcomes

When you run this architecture, your logs will show the system attempting your primary provider first. If OpenAI hits a 500 server error or a 429 rate limit, the loop catches the exception in milliseconds and switches to Claude or Gemini.

The calling application receives a standard LLMResponse regardless of which vendor actually processed the request. Your downstream data parsing remains identical, keeping your application stable even during major cloud vendor outages.

If you need help scaling this pattern across multi-agent systems or enterprise infrastructure, the team at Gaper provides senior engineers who specialize in production ready AI systems.

How to Manage Long-Running State and Asynchronous Callbacks in Autonomous Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:51:10 +0000

Building autonomous agents that handle simple, single-turn prompts is easy. The real headache starts when your agent has to wait three days for a human manager to approve an action, or ten minutes for an external API to process a batch of files. If your agent holds its context in memory while waiting, a single server restart wipes out everything. I learned this the hard way when a deployment killed dozens of active workflows mid-execution, leaving users stranded with half-finished tasks.

To build reliable long running agents, you need a system that decouples execution from memory. Your agent must be able to pause, write its full context to storage, yield control, and wake up only when a trigger or callback arrives.

Here is how you can implement durable agent state management and asynchronous callbacks step by step.

Traditional agent scripts run in a continuous loop: think, act, observe, repeat. This works fine for quick scripts, but fails for long-running processes.

[ Traditional Agent Loop ]
Agent Start -> Prompt LLM -> Call Tool -> Wait for API (Blocking) -> Process -> Done
(If process crashes during Wait, state is lost forever)

[ Event-Driven Durable Agent ]
Agent Start -> Prompt LLM -> Save State to DB -> Suspend Process
                                                        |
   Webhook / Human Input arrives later -----------------+
                                                        v
                               Resume Process -> Load State -> Complete Task

To handle asynchronous callbacks, we break this single process into three main parts:

State Persistence: Saving the memory, scratchpad, and conversation history to a database.
Execution Suspension: Halting execution cleanly without holding open HTTP connections or thread locks.
Event Rehydration: Rebuilding the agent's full state when a webhook or user input triggers a wake-up call.

If you are scaling complex AI agent development, getting this foundation right early saves hundreds of hours of debugging downstream.

Step 1: Define a Serializable Agent State Schema

You cannot save raw LLM instances or active client connections to a database. You must serialize the agent state into a structured schema containing the prompt history, tool call outputs, pending callbacks, and current status.

Here is a Pydantic schema using Python that captures what a durable execution setup needs:

from enum import Enum
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
import datetime

class AgentStatus(str, Enum):
    RUNNING = "running"
    SUSPENDED = "suspended"
    COMPLETED = "completed"
    FAILED = "failed"

class AgentState(BaseModel):
    thread_id: str
    status: AgentStatus = AgentStatus.RUNNING
    system_prompt: str
    messages: List[Dict[str, Any]] = Field(default_factory=list)
    pending_callback_id: Optional[str] = None
    context_data: Dict[str, Any] = Field(default_factory=dict)
    created_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())
    updated_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())

This model stores everything necessary to rebuild the agent environment. The pending_callback_id field tracks external async actions like email responses, third-party webhook processing, or human approvals.

Step 2: Implement Persistent Storage for Agent State Management

Next, build a persistence repository. SQLite works great for local testing, while PostgreSQL or Redis fits production workloads.

This simple repository pattern manages durable state transitions:

import sqlite3
import json

class StateRepository:
    def __init__(self, db_path: str = "agent_state.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS states (
                    thread_id TEXT PRIMARY KEY,
                    status TEXT,
                    pending_callback_id TEXT,
                    payload TEXT
                )
            """)
            conn.commit()

    def save(self, state: AgentState):
        state.updated_at = datetime.datetime.utcnow().isoformat()
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                INSERT INTO states (thread_id, status, pending_callback_id, payload)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(thread_id) DO UPDATE SET
                    status=excluded.status,
                    pending_callback_id=excluded.pending_callback_id,
                    payload=excluded.payload
                """,
                (
                    state.thread_id,
                    state.status.value,
                    state.pending_callback_id,
                    json.dumps(state.dict())
                )
            )
            conn.commit()

    def load(self, thread_id: str) -> Optional[AgentState]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT payload FROM states WHERE thread_id = ?", (thread_id,))
            row = cursor.fetchone()
            if row:
                return AgentState(**json.loads(row[0]))
            return None

With persistent storage ready, the agent can stop execution entirely while waiting for asynchronous events.

Step 3: Handle Asynchronous Callbacks and Suspensions

When your agent hits an async barrier, such as waiting for a human review or calling a long-running batch extraction API, it should emit an external request, register a unique callback key, save state, and exit.

Here is an example showing how an agent suspends its flow:

import uuid

class AgentEngine:
    def __init__(self, repo: StateRepository):
        self.repo = repo

    def execute_human_approval_step(self, thread_id: str, action_details: str):
        state = self.repo.load(thread_id)
        if not state:
            raise ValueError("Thread not found")

        # Create a unique callback listener token
        callback_id = f"cb_{uuid.uuid4().hex[:8]}"

        # Record action intent into conversation history
        state.messages.append({
            "role": "assistant",
            "content": f"Requesting human approval for action: {action_details}"
        })

        # Transition state to suspended
        state.status = AgentStatus.SUSPENDED
        state.pending_callback_id = callback_id

        # Save state to storage
        self.repo.save(state)

        print(f"[Agent Suspended] Task waiting for callback token: {callback_id}")
        return callback_id

Instead of blocking thread execution, this design frees memory and CPU resources. It is especially useful when integrating complex backend tasks with workflow automation pipelines that spans across external services.

Step 4: Resume Execution on Callback Arrival

Now, set up a handler to accept incoming callbacks, find the matching agent execution state, inject the incoming payload, and resume processing.

    def handle_callback(self, callback_id: str, payload: Dict[str, Any]):
        # Search state database for matching callback token
        with sqlite3.connect(self.repo.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT thread_id FROM states WHERE pending_callback_id = ?", (callback_id,))
            row = cursor.fetchone()

        if not row:
            raise ValueError(f"No agent suspended for callback ID: {callback_id}")

        thread_id = row[0]
        state = self.repo.load(thread_id)

        # Update thread state with incoming payload
        state.messages.append({
            "role": "user",
            "content": f"Callback received with data: {json.dumps(payload)}"
        })

        # Clear callback lock and set back to running
        state.pending_callback_id = None
        state.status = AgentStatus.RUNNING
        self.repo.save(state)

        print(f"[Agent Resumed] Processing thread: {thread_id}")
        return self.continue_agent_loop(state)

    def continue_agent_loop(self, state: AgentState):
        # Continue reasoning loop with updated state context
        state.messages.append({
            "role": "assistant",
            "content": "Received callback approval. Proceeding with execution."
        })
        state.status = AgentStatus.COMPLETED
        self.repo.save(state)
        return state

Step 5: Put It Together and Run an Simulation

Here is a full demonstration of the cycle: starting an agent task, pausing for external input, and resuming execution.

if __name__ == "__main__":
    repo = StateRepository()
    engine = AgentEngine(repo)

    # Initialize a new agent run
    thread_id = f"session_{uuid.uuid4().hex[:6]}"
    initial_state = AgentState(
        thread_id=thread_id,
        system_prompt="You are an autonomous assistant handling invoice operations.",
        messages=[{"role": "user", "content": "Process refund for Order #9482"}]
    )
    repo.save(initial_state)

    # Agent runs until it needs external authorization
    callback_token = engine.execute_human_approval_step(
        thread_id=thread_id, 
        action_details="Refund $450 to customer"
    )

    # At this point, the application process can close, restart, or crash.
    # The agent state remains safe inside SQLite.

    # Later, an incoming webhook or API call fires this endpoint handler:
    print("\n... Waiting for external webhook callback ...\n")

    incoming_webhook_payload = {
        "approved": True,
        "approver": "manager@company.com"
    }

    final_state = engine.handle_callback(callback_token, incoming_webhook_payload)
    print(f"Final Agent Status: {final_state.status.value}")

Running this code prints:

[Agent Suspended] Task waiting for callback token: cb_a1b2c3d4

... Waiting for external webhook callback ...

[Agent Resumed] Processing thread: session_e5f6g7
Final Agent Status: completed

Best Practices for Production Durable Execution

When building production grade systems for async agent workflows, keep these tips in mind:

Use Dedicated Task Orchestrators: While SQLite works for basic usage, engines like Temporal, Prefect, or AWS Step Functions handle re-runs, retries, and time-outs much better.
Store Minimal Payload Context: Do not save large binary outputs directly inside the agent execution payload. Store them in S3 or object storage, then pass references into the agent state object.
Handle Timeouts: Always append expiration timers to suspended states. If a callback never returns, a fallback background job should wake the agent up to handle the error condition gracefully.

Building long running agents requires shifting from simple synchronous loops to robust, stateful architectures. If your team needs hand building reliable agent infrastructures, our engineers at Gaper can help design state management patterns suited for enterprise scale.

Why Fine-Tuning Isn't Enough: Combining RAG, Tools, and Runtime Guardrails for Production Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:48:13 +0000

Last year, I watched an engineering team spend six months and tens of thousands of dollars fine-tuning an open-source model on their company's internal documentation. They were convinced it would replace their front-line customer support stack. On launch day, the bot gave polite, beautifully formatted answers that were completely, spectacularly wrong. It hallucinated pricing plans that never existed and cited policy documents from three years ago.

That project was scrapped two weeks later, but the lesson stuck with me. We are watching hundreds of teams make the exact same mistake right now. They treat weight training like a magic database.

Fine tuning is great for style, tone, and syntax. If you want a model to sound like a medical professional, adhere to strict JSON formats, or master a rare programming syntax, fine tuning works well.

Fine tuning is not memory. It is a static snapshot.

The moment you finish training a model checkpoint, your data starts decaying. When your product team updates pricing tomorrow, your shiny fine-tuned model becomes an operational hazard. Re-training every time your business logic changes is slow, expensive, and fragile. You end up fighting catastrophic forgetting, where the model forgets general reasoning skills just to memorize a few API endpoints.

The Real Architecture of Production AI Agents

Building reliable production AI agents requires a modular software architecture. You cannot rely on a single neural network to act as database, logic engine, API controller, and security guard all at once. Basic software engineering rules still apply here. We need clear separation of concerns.

A production agent system rests on three distinct pillars that sit outside the base model weights: Retrieval-Augmented Generation (RAG), external tool integration, and runtime guardrails.

RAG and Tools: Giving the Model Sight and Motion

Think of a large language model as an intelligent consultant with temporary amnesia. RAG is the folder of fresh background context you place on their desk before asking a question. By decoupling knowledge from model weights, you can update your vector stores and knowledge graphs instantly. When a policy changes, you update your database, not your training scripts.

Tools give the agent hands. An LLM should never calculate loan amortization or query a database through raw token prediction. It will fail. Instead, the model should decide which tool to call, pass structured parameters, and let a deterministic script do the real work.

When you need complex business logic or enterprise workflow automation, tools let the model orchestrate existing, battle-tested software rather than reinventing execution logic on the fly.

Runtime Guardrails Are Not Optional

Even with pristine context and reliable tools, models drift. They get confused by long prompt histories. They attempt invalid function calls. They break character when a user tries a basic prompt injection.

This is where runtime guardrails come in. You need deterministic software layers wrapping your model calls.

Input guardrails inspect incoming user prompts for malicious intent, topical relevance, and Pydantic schema alignment before the model ever sees them. Output guardrails parse the raw text response, check for policy violations, validate output structures, and verify facts against the original retrieved context. If the model strays, the system catches it, logs the anomaly, and triggers a retry loop or a safe fallback mechanism.

Fine tuning might take your output format quality from 70% to 90%. Runtime guardrails and schema validation are what take your system reliability from 90% to 99.9%.

Where Fine Tuning Actually Belongs

Fine tuning isn't dead, but its role in the stack has shifted.

Instead of using fine tuning to teach your model facts, use it to lower latency and cut token overhead. If your system prompt requires 2,000 tokens of few-shot examples just to force a specific output schema, that is a great candidate for fine tuning. Train a smaller, faster model on that specific format, then put that model inside a runtime loop powered by dynamic RAG, structured tool calling, and guardrails.

Stop expecting a single weight file to solve architectural problems. Build robust systems around your models instead.

If you want to build durable systems without spending months down the wrong architectural path, take a look at the team at Gaper. Their engineers specialize in building production-grade agent frameworks and providing practical AI consulting to help companies deploy models that actually work in the real world.

Designing Scoped API Authorization and RBAC for Autonomous Support Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:44:34 +0000

Autonomous support agents need access to internal APIs to issue refunds, update accounts, and retrieve user logs. However, giving an AI model direct access to your database or an admin API key poses a massive risk. If a user manipulates the agent through prompt injection, the agent could wipe a database or issue unauthorized payments. To prevent this, you need a security framework that restricts what an agent can do based on the authenticated user, the agent's assigned role, and strict API scopes.

Before building your authorization layer, make sure you have:

An API gateway or backend service running Node.js, Python, or Go.
An OAuth 2.0 Identity Provider supporting RFC 8693 token exchange.
An LLM framework set up for tool use or function calling.
Basic understanding of RBAC and JSON Web Tokens (JWTs).

Step 1: Define Scopes for Every Agent Tool

Never expose generic API endpoints to your AI tools. Instead, break your backend capabilities into explicit, low-privilege scopes.

Instead of giving your agent access to a generic POST /api/v1/payments route, create dedicated tool abstractions with specific scopes:

read:orders: View order history.
read:user_profile: View basic customer info.
write:refund_limited: Issue refunds up to a specific threshold, like $50.

Map these scopes directly to the tools in your agent's system prompt or function definitions. If the agent does not need to edit a user profile, its token context should not even possess the scope to do so.

Step 2: Implement OAuth Delegation for Agent Contexts

An agent should never act with full system admin rights. Use token delegation so the agent inherits the permissions of the user requesting help, constrained by the agent's own role.

When a customer starts a chat session:

Authenticate the customer and issue a short lived user token.
Request a delegated token for the support agent service using token exchange (RFC 8693).
The returned token contains the intersection of user permissions and agent permissions.

If a non-admin user asks the agent to delete an account, the delegated token won't have the delete:account scope. Even if prompt injection succeeds, the API gateway will block the tool call with a 403 Forbidden. If you are building complex integration pipelines, setting up clean workflow automation patterns helps keep token exchanges reliable across services.

Step 3: Add API Gateway Tool Authorization Middleware

Validate incoming tool execution requests at your API gateway or middleware, not inside the LLM logic. Treat the agent as an untrusted client.

Here is a simple Python example using FastAPI to enforce scope checks on a tool execution endpoint:

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

app = FastAPI()
security = HTTPBearer()

def verify_agent_scope(required_scope: str):
    def dependency(credentials: HTTPAuthorizationCredentials = Security(security)):
        token = credentials.credentials
        try:
            payload = jwt.decode(token, "YOUR_SECRET_KEY", algorithms=["HS256"])
            scopes = payload.get("scopes", []).split(" ")
            if required_scope not in scopes:
                raise HTTPException(status_code=403, detail="Insufficient tool permissions")
            return payload
        except jwt.PyJWTError:
            raise HTTPException(status_code=401, detail="Invalid token")
    return dependency

@app.post("/tools/issue-refund")
def issue_refund(amount: float, payload: dict = Depends(verify_agent_scope("write:refund_limited"))):
    if amount > 50.0:
        raise HTTPException(status_code=400, detail="Amount exceeds automated limit")
    return {"status": "success", "refunded": amount}

This ensures that even if the agent hallucinates a function call with invalid parameters, your backend enforces hard security limits.

Step 4: Require Step-Up Authorization for High-Risk Operations

For operations that exceed normal bounds, require step-up authorization.

If a customer asks for a $200 refund, the agent tool should return an approval_required status message instead of executing the API call directly. The backend generates an approval task sent to a human support queue. Once the human approves, a temporary, single-use token is generated to complete the transaction.

If you want to skip building these RBAC architectures from scratch, custom AI agent development services can help structure secure tool infrastructure right out of the box.

Expected Outcomes

Once you implement this RBAC design, you can expect:

Strong defense against prompt injection attacks that attempt unauthorized tool execution.
Granular audit logs showing which user session requested an action and which agent context executed it.
Reduced operational risk by enforcing strict value ceilings on autonomous tasks.

If you are looking for guidance on implementing secure agent infrastructure or scaling custom models, check out Gaper for technical resources and engineering support.

How to Build an Immutable Audit Logging Pipeline for AI Agent Actions

MUHAMMAD MUSADIQ — Tue, 28 Jul 2026 05:27:58 +0000

title: How to Build an Immutable Audit Logging Pipeline for AI Agent Actions
published: false

description: Standard logs fall short when tracking autonomous AI agents. Learn how to build a tamper-proof, cryptographically sealed audit logging pipeline to guarantee compliance and securely record every agent action.

When an autonomous AI agent executes a bad database query or calls an external API with unexpected parameters, figuring out what happened is a nightmare. Standard application logs quickly fall short. They can be edited, rotated, or deleted, which breaks trust and makes AI compliance nearly impossible. If your agents handle real-world operations, you need tamper-proof audit logs that guarantee every step of an agent's execution is recorded, linked, and cryptographically sealed.

I learned this the hard way when an experimental customer support agent started firing off refund requests. Without a clean event history, I spent hours untangling state changes. In this guide, we will build a production-grade, immutable audit logging pipeline using Python, cryptographic hash chains, and append-only storage concepts.

An immutable audit trail needs three main features:

Structured payloads capturing inputs, outputs, tool calls, and state metadata.
Cryptographic hashing where each log entry contains the hash of the preceding entry, forming a tamper-evident chain.
Append-only storage that prevents modification or deletion of past records.

Here is how data flows through our observability pipeline:

[ Agent Action / Tool Call ] 
          │
          ▼
[ Log Interceptor / Context ]
          │
          ▼
[ Hash Chaining Engine ] (SHA-256 link to previous entry)
          │
          ▼
[ Append-Only Storage ] (PostgreSQL / S3 Object Lock)

Let's write the code step by step.

Step 1: Define the Log Entry and Cryptographic Engine

First, create a structured log entry schema. We will use standard SHA-256 hashes to link log entries together. If anyone alters an old log entry, the entire hash chain breaks, immediately signaling tampering.

Create a file named audit_chain.py:

import hashlib
import json
from datetime import datetime, timezone
from typing import Dict, Any, Optional

class AuditEntry:
    def __init__(self, agent_id: str, action: str, payload: Dict[str, Any], previous_hash: str = "0"):
        self.timestamp = datetime.now(timezone.utc).isoformat()
        self.agent_id = agent_id
        self.action = action
        self.payload = payload
        self.previous_hash = previous_hash
        self.hash = self.calculate_hash()

    def calculate_hash(self) -> str:
        # Create a deterministic string representation of the entry data
        data = {
            "timestamp": self.timestamp,
            "agent_id": self.agent_id,
            "action": self.action,
            "payload": self.payload,
            "previous_hash": self.previous_hash
        }
        serialized = json.dumps(data, sort_keys=True)
        return hashlib.sha256(serialized.encode('utf-8')).hexdigest()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "timestamp": self.timestamp,
            "agent_id": self.agent_id,
            "action": self.action,
            "payload": self.payload,
            "previous_hash": self.previous_hash,
            "hash": self.hash
        }

This structure ensures that every entry binds itself to the previous state.

Step 2: Intercept Agent Execution

To record agent tracing data without cluttering your core agent logic, wrap tool calls and LLM invocations using Python decorators or middleware.

If you are building custom agents for enterprise workflows like those supported by Gaper's AI Agent Development services, keeping tracing decoupled from core business logic is essential.

Create logger.py to maintain the chain state in memory before writing to disk or database:

import json
import os
from audit_chain import AuditEntry

class AuditLogger:
    def __init__(self, log_file: str = "audit_log.json"):
        self.log_file = log_file
        self.last_hash = self._get_last_hash()

    def _get_last_hash(self) -> str:
        if not os.path.exists(self.log_file):
            return "0" * 64

        with open(self.log_file, "r") as f:
            lines = f.readlines()
            if not lines:
                return "0" * 64
            last_line = lines[-1]
            entry = json.loads(last_line)
            return entry["hash"]

    def log_action(self, agent_id: str, action: str, payload: dict) -> dict:
        entry = AuditEntry(
            agent_id=agent_id,
            action=action,
            payload=payload,
            previous_hash=self.last_hash
        )

        # Persist to append-only log file
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry.to_dict()) + "\n")

        self.last_hash = entry.hash
        return entry.to_dict()

Now, let's create a decorator to wrap agent functions automatically:

from functools import wraps

audit_logger = AuditLogger()

def trace_agent_action(agent_id: str, action_name: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            payload = {
                "inputs": {"args": args, "kwargs": kwargs}
            }
            try:
                result = func(*args, **kwargs)
                payload["output"] = result
                payload["status"] = "SUCCESS"
                audit_logger.log_action(agent_id, action_name, payload)
                return result
            except Exception as e:
                payload["error"] = str(e)
                payload["status"] = "FAILED"
                audit_logger.log_action(agent_id, action_name, payload)
                raise e
        return wrapper
    return decorator

Step 3: Execute Agent Actions and Log Events

Now let's simulate an AI agent performing actions. This could be a financial bot processing an invoice or an automated assistant operating in regulated fields, where AI in accounting and finance requires complete transparency and verifiable audit logs.

Create agent.py:

from logger import trace_agent_action

AGENT_ID = "finance_agent_v1"

@trace_agent_action(agent_id=AGENT_ID, action_name="fetch_user_balance")
def fetch_user_balance(user_id: str) -> float:
    # Simulated API call
    return 1450.50

@trace_agent_action(agent_id=AGENT_ID, action_name="execute_transfer")
def execute_transfer(user_id: str, target_account: str, amount: float) -> str:
    balance = fetch_user_balance(user_id)
    if amount > balance:
        raise ValueError("Insufficient funds")

    # Simulated execution
    return f"Transferred ${amount} to {target_account}"

if __name__ == "__main__":
    print("Running agent tasks...")
    execute_transfer("usr_99", "acc_441", 200.00)

    try:
        execute_transfer("usr_99", "acc_441", 5000.00)
    except Exception as e:
        print(f"Handled expected error: {e}")

Run python agent.py. This will generate audit_log.json containing JSON Lines data where each entry points to the hash of the entry before it.

Step 4: Verify Audit Log Integrity

The primary benefit of a cryptographic chain is simple verification. If an attacker or corrupt process alters any byte in audit_log.json, the hashes will fail to match downstream.

Create a verification script verify_chain.py:

import json
import hashlib

def verify_audit_log(file_path: str) -> bool:
    if not os.path.exists(file_path):
        print("No audit log found.")
        return False

    with open(file_path, "r") as f:
        lines = f.readlines()

    expected_previous_hash = "0" * 64

    for i, line in enumerate(lines):
        entry = json.loads(line)

        # 1. Check if previous_hash matches expected hash from previous iteration
        if entry["previous_hash"] != expected_previous_hash:
            print(f" Integrity failure at line {i+1}: Previous hash mismatch!")
            return False

        # 2. Recalculate hash of current entry
        data_to_hash = {
            "timestamp": entry["timestamp"],
            "agent_id": entry["agent_id"],
            "action": entry["action"],
            "payload": entry["payload"],
            "previous_hash": entry["previous_hash"]
        }
        recalculated_hash = hashlib.sha256(
            json.dumps(data_to_hash, sort_keys=True).encode('utf-8')
        ).hexdigest()

        if recalculated_hash != entry["hash"]:
            print(f" Integrity failure at line {i+1}: Hash content recalculation failed!")
            return False

        expected_previous_hash = entry["hash"]

    print(" Audit log integrity verified. No tampering detected.")
    return True

if __name__ == "__main__":
    verify_audit_log("audit_log.json")

To test verification, run python verify_chain.py. You will see a success message. If you manually open audit_log.json and change a dollar amount or timestamp, re-running the script will instantly report a chain break.

Hardening Storage for Production

Writing to a local JSON file works for prototyping, but production observability systems demand stricter protections:

AWS S3 Object Lock: Sync log batches to S3 buckets configured with Write-Once-Read-Many (WORM) storage in Compliance Mode. This prevents root users from deleting or altering log objects for a defined retention period.
Database Triggers: If using PostgreSQL, write log entries to an append-only table. You can revoke UPDATE and DELETE permissions for the application DB user, allowing only INSERT privileges.
Remote Log Aggregation: Ship logs off the host machine instantly via vector agents or streaming queues like Kafka to reduce local surface area for log tampering.

Summary

Building an immutable pipeline gives you immediate visibility into agent execution, making debugging straightforward while satisfying strict audit logs mandates. Cryptographic chaining guarantees that your execution history remains trustworthy, even when scaling complex multi-agent workflows.

If you need help building custom AI solutions, scaling agent infrastructure, or sourcing seasoned engineers to build compliant pipelines, check out Gaper for expert technical support and engineering talent.

How to Set Up Synthetic Evals for Non-Deterministic Support Agent Workflows

MUHAMMAD MUSADIQ — Tue, 28 Jul 2026 05:16:50 +0000

Testing AI support agents is tricky because they rarely produce the exact same output twice. Traditional unit tests break immediately when applied to non-deterministic workflows, such as customer refund requests, order updates, or troubleshooting steps. By generating synthetic data to simulate edge-case conversations and evaluating the agent against predefined assertions, you can build reliable pass-rate tests and production guardrails without waiting for real users to break your system.

Python 3.10 or higher installed
An OpenAI API key or access to an alternative LLM provider
A dataset framework or library like Promptfoo, Ragas, or custom Python scripts
Basic understanding of JSON schemas and function calling

Step 1: Generate Synthetic Personas and Test Cases

Real user logs are useful, but they rarely cover every edge case. You need a separate script to generate synthetic user personas with varying tones, misspellings, and incomplete information.

Create a script named generate_synthetic_data.py to produce a structured JSON file of test cases:

import json
from openai import OpenAI

client = OpenAI()

system_prompt = """
You are an expert QA engineer creating test cases for a customer support AI.
Generate 5 diverse customer queries for a return policy workflow.
Include varying emotional states, typos, missing order numbers, and edge cases.
Return a valid JSON array of objects with keys: "id", "user_message", "expected_intent", "should_escalate".
"""

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": system_prompt}],
    response_format={"type": "json_object"}
)

with open("synthetic_tests.json", "w") as f:
    f.write(response.choices[0].message.content)

print("Synthetic test cases generated successfully.")

Run this script to create your initial eval dataset.

Step 2: Define Assertions and Evaluator Metrics

Exact string matching will not work here. Instead, write an LLM-as-a-judge evaluation function that reviews the support agent's output against your criteria.

If you are working on custom AI agent development, focus on three main metrics: policy adherence, tone, and tool choice accuracy.

Create a judge function in eval_judge.py:

def evaluate_response(user_query, agent_response, expected_intent):
    judge_prompt = f"""
    Evaluate the following support agent response based on the customer query.

    Customer Query: {user_query}
    Expected Intent: {expected_intent}
    Agent Response: {agent_response}

    Criteria:
    1. Did the agent address the query accurately according to standard return rules?
    2. Was the tone professional and helpful?
    3. Did it avoid making up fake policies?

    Reply with 'PASS' if all criteria are met, or 'FAIL' followed by a short reason.
    """

    res = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": judge_prompt}]
    )

    output = res.choices[0].message.content.strip()
    return "PASS" in output, output

Step 3: Run Pass Rate Testing Across Iterations

Run your support agent through all synthetic test cases and calculate the baseline pass rate.

import json

with open("synthetic_tests.json", "r") as f:
    data = json.load(f)
    test_cases = data.get("test_cases", data)

passed = 0
total = len(test_cases)

for test in test_cases:
    # Replace this with your actual support agent function call
    agent_output = "I can help with that return! Please provide your order ID." 

    is_pass, reason = evaluate_response(test["user_message"], agent_output, test["expected_intent"])
    if is_pass:
        passed += 1
    else:
        print(f"Failed Test {test['id']}: {reason}")

pass_rate = (passed / total) * 100
print(f"\nOverall Pass Rate: {pass_rate:.1f}%")

Whenever you update your system prompts or retrieval context, re-run this test suite. If the pass rate drops below your benchmark (for example, 90%), block the deployment.

Step 4: Hook Evals into Production Guardrails

Synthetic evals should not live only in your test environment. Run light, fast assertions directly inside your live workflow automation setup.

If a live response fails critical assertions (such as detecting toxic language or hallucinatory refund promises), interrupt the execution chain immediately and route the user to a human representative.

Expected Outcomes

After completing this setup, you can expect:

A repeatable methodology to score non-deterministic responses mathematically.
Clear pass rate metrics (e.g., target 92% accuracy across 100+ synthetic scenarios).
Prevention of prompt regressions before code hits production.
Real-time guardrails that prevent rogue LLM outputs from reaching actual customers.

If you need help building reliable AI pipelines or scaling custom agents for your business, the developers at Gaper can help you architect and deploy production-grade systems tailored to your technical stack.

How to Build Idempotent Tool Wrappers to Prevent Duplicate AI Agent Actions

MUHAMMAD MUSADIQ — Tue, 28 Jul 2026 05:12:00 +0000

AI agents often retry tool calls when an API connection drops or when the model gets confused by a slow response. If your agent is processing credit card refunds, sending customer emails, or updating a database, these repeated calls can create massive headaches. I once watched a buggy agent send five duplicate confirmation emails to a high-value client because of a network timeout. You can prevent this entirely by wrapping your agent's tools in an idempotency layer that catches duplicate requests before they touch your core services.

Before starting, make sure you have the following ready:

Python 3.10 or newer installed on your machine
Redis installed locally or an active Redis Cloud instance (a standard Python dictionary works for simple local testing)
Basic familiarity with Python decorators and JSON handling

Step 1: Generate Deterministic Idempotency Keys

An operation is idempotent if running it multiple times produces the exact same result as running it once. To achieve this, your wrapper needs a way to uniquely identify every incoming request.

You can derive a key automatically by hashing the tool name along with its input arguments. Here is a simple function to build a deterministic key:

import hashlib
import json
from typing import Any, Dict

def generate_tool_key(tool_name: str, kwargs: Dict[str, Any]) -> str:
    # Sort keys so {"a": 1, "b": 2} matches {"b": 2, "a": 1}
    serialized_args = json.dumps(kwargs, sort_keys=True)
    raw_string = f"{tool_name}:{serialized_args}"
    return hashlib.sha256(raw_string.encode('utf-8')).hexdigest()

If the agent sends send_email(recipient="test@example.com", body="Hello") three times in a row, this function generates the exact same hash every single time.

Step 2: Build the Idempotency Wrapper Decorator

Next, write a decorator that intercepts tool calls before execution. The wrapper checks your cache store for the generated key. If the key exists, it returns the previously saved output instantly without calling the backend API again.

Here is a full Python implementation using a simple local store:

import functools
import time

# Simple in-memory cache for demonstration; replace with Redis in production
CACHE_STORE = {}

def idempotent_tool(ttl_seconds: int = 3600):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            tool_name = func.__name__
            cache_key = generate_tool_key(tool_name, kwargs)

            # Check if we already processed this exact action
            if cache_key in CACHE_STORE:
                cached_entry = CACHE_STORE[cache_key]
                # Check if cache is still valid
                if time.time() - cached_entry['timestamp'] < ttl_seconds:
                    print(f"[IDEMPOTENT] Returning cached response for {tool_name}")
                    return cached_entry['result']

            # Execute the actual tool function if key is not cached
            result = func(*args, **kwargs)

            # Store the result in cache
            CACHE_STORE[cache_key] = {
                'result': result,
                'timestamp': time.time()
            }
            return result
        return wrapper
    return decorator

If you work on complex backend tasks like workflow automation, replacing the in-memory dictionary with Redis ensures that key checks persist across distributed worker nodes.

Step 3: Wrap Your Agent Functions

Now apply the @idempotent_tool() decorator to your tool definitions. Here is an example of an external payment call wrapped for safe agent usage:

@idempotent_tool(ttl_seconds=86400)
def process_refund(user_id: str, amount: float, reason: str) -> dict:
    # Simulate an expensive or risky API call
    print(f"[API CALL] Processing refund of ${amount} for user {user_id}...")

    # Fake successful payment gateway payload
    return {
        "status": "success",
        "transaction_id": f"tx_rf_{user_id}_9981",
        "amount_refunded": amount
    }

Step 4: Test the Wrapper Against Agent Retries

Simulate an agent hitting an API timeout loop by calling the wrapped function multiple times back-to-back with identical arguments.

if __name__ == "__main__":
    payload = {
        "user_id": "usr_4021",
        "amount": 49.99,
        "reason": "Accidental double charge"
    }

    print("--- First Call (Agent initiates action) ---")
    res1 = process_refund(**payload)
    print("Result 1:", res1)

    print("\n--- Second Call (Agent retries due to local timeout) ---")
    res2 = process_refund(**payload)
    print("Result 2:", res2)

Expected Output

When you run the script, the first call executes the underlying function and hits the mock payment gateway. The second call detects the matching request signature, skips the API code entirely, and returns the cached payload.

--- First Call (Agent initiates action) ---
[API CALL] Processing refund of $49.99 for user usr_4021...
Result 1: {'status': 'success', 'transaction_id': 'tx_rf_usr_4021_9981', 'amount_refunded': 49.99}

--- Second Call (Agent retries due to local timeout) ---
[IDEMPOTENT] Returning cached response for process_refund
Result 2: {'status': 'success', 'transaction_id': 'tx_rf_usr_4021_9981', 'amount_refunded': 49.99}

The underlying payment API only ran once. Your user was not double refunded, and your agent received the expected response object without crashing.

Managing Edge Cases and Locks

In production setups, you need to handle in-flight requests. If an agent fires off two duplicate network requests within milliseconds of each other, both might check the cache before either finishes writing to it.

To solve this, set a short lock key (like IN_PROGRESS) in Redis as soon as a tool call begins. If a second request arrives while that lock key exists, force the second call to wait a few seconds and check the cache again instead of executing the underlying logic right away.

For critical production workloads, implementing robust tool safety patterns is essential. If you want experienced engineers to build or audit your agent infrastructure, Gaper can connect you with vetted, remote software developers who specialize in custom AI Agent Development.

How to Implement Model Context Protocol (MCP) for Custom AI Agent Tooling

MUHAMMAD MUSADIQ — Tue, 28 Jul 2026 05:07:19 +0000

Building tools for AI agents used to feel like reinventing the wheel every week. You wrote custom wrappers, handcrafted JSON schemas, managed token limits, and prayed the model wouldn't hallucinate arguments. Then, Anthropic open-sourced the Model Context Protocol (MCP), giving developers a clean standard for connecting language models to local files, databases, and custom APIs.

In this guide, I will show you how to build and run a custom MCP server from scratch to power your own agent integrations.

At its core, mcp acts like a USB port for LLMs. Instead of hardcoding custom API logic directly into your host application, your MCP server exposes tools, resources, and prompt templates over a standard protocol. The client discovers these tools dynamically.

When a user interacts with the system, the client sends available tool descriptions to the LLM. The LLM decides which tool to execute and returns a structured tool call request. Your MCP server executes the function and returns the result back to the model. This decouples your underlying application logic from your llm deployment strategy.

If you are designing high throughput software pipelines, reading up on modern ai agent development patterns will help you plan your architecture effectively.

Step 1: Set Up Your Project Environment

We will build an MCP server in Node.js using TypeScript and the official MCP SDK.

Create a new directory and initialize the Node package:

mkdir mcp-custom-tools
cd mcp-custom-tools
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node tsx
npx tsc --init

Ensure your tsconfig.json targets a modern Node environment. Setting "target": "ES2022" and "moduleResolution": "NodeNext" works cleanly.

Step 2: Define Your MCP Tools

Create a file named index.ts. We will define a simple tool that simulates looking up a user record and triggering an external api integration.

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

const server = new Server(
  {
    name: "user-management-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// Define arguments schema using Zod
const GetUserArgsSchema = z.object({
  userId: z.string().describe("The unique ID of the user"),
});

// Register tool definitions
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "get_user_profile",
        description: "Fetch user profile details from the internal database by user ID.",
        inputSchema: {
          type: "object",
          properties: {
            userId: {
              type: "string",
              description: "The unique ID of the user",
            },
          },
          required: ["userId"],
        },
      },
    ],
  };
});

Step 3: Implement Tool Execution Logic

Now, handle the incoming tool calling requests inside the server logic.

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "get_user_profile") {
    const { userId } = GetUserArgsSchema.parse(request.params.arguments);

    // Mock API call or database query
    const userDatabase: Record<string, { name: string; role: string; email: string }> = {
      "usr-101": { name: "Sarah Connor", role: "Admin", email: "sarah@example.com" },
      "usr-102": { name: "Kyle Reese", role: "Engineer", email: "kyle@example.com" },
    };

    const user = userDatabase[userId];

    if (!user) {
      return {
        isError: true,
        content: [
          {
            type: "text",
            text: `User with ID ${userId} was not found.`,
          },
        ],
      };
    }

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(user, null, 2),
        },
      ],
    };
  }

  throw new Error(`Unknown tool: ${request.params.name}`);
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server listening on stdio interface");
}

main().catch((err) => {
  console.error("Fatal error:", err);
  process.exit(1);
});

Step 4: Test Your MCP Server

You can connect your new server directly to Claude Desktop or any client supporting the standard protocol.

Open your local configuration file (for Claude Desktop, this is typically located at ~/Library/Application Support/Claude/claude_desktop_config.json on macOS):

{
  "mcpServers": {
    "user-management": {
      "command": "npx",
      "args": [
        "tsx",
        "/absolute/path/to/mcp-custom-tools/index.ts"
      ]
    }
  }
}

Restart your host client. When you ask the agent "Can you check profile details for usr-101?", the client recognizes the tool definition, formats the JSON-RPC call over standard input/output, and renders the user details in the conversation.

As your tool count grows, combining MCP servers with enterprise workflow automation logic will help you coordinate actions across disparate backends cleanly.

Real Lessons Learned From Building MCP Tools

I ran into a few subtle issues when deploying my first set of custom servers. Keep these tips in mind:

Never log using standard output. STDIO transport relies strictly on stdout for standard JSON-RPC framing. If you write console.log() anywhere, you break the parser. Always write logs to console.error().
Keep tool descriptions explicit. Models rely heavily on the description field to determine when to trigger tool calls. Give concrete examples in the text if the model skips your tools.
Keep individual tool boundaries small. A single tool should perform one specific operation. Do not build mega-tools that try to handle three distinct workflows at once.

Building robust custom infrastructure takes time. If you need dedicated engineers to help build out your agent workflows, Gaper connects companies with vetted software developers specialized in AI engineering.

Build vs. Buy AI Agents: An Architectural Framework for Engineering Leaders

MUHAMMAD MUSADIQ — Mon, 27 Jul 2026 18:34:13 +0000

Last month, I watched a VP of Engineering burn six figures on a shiny enterprise AI agent platform. It promised zero-code automation for customer support and internal ticket routing. Three months into the rollout, the engineering team quietly turned it off. Nobody could figure out why the agent kept losing context mid-session or hallucinating bad database queries against production systems. The slick dashboard masked a nightmare of brittle prompt wrappers that no developer could debug.

This story keeps happening. Engineering leaders are under massive pressure from executives to show immediate AI wins. The easy path looks like buying a prepackaged platform that promises turnkey AI workers. But in practice, buying generic SaaS agent platforms for core engineering workflows is a trap. You end up spending more time wrestling with vendor guardrails and opaque abstractions than it takes to write clean code.

The real issue comes down to AI architecture. An AI agent is not just a call to an OpenAI API with a fancy system prompt. It is a stateful system. It needs long term memory, deterministic fallback mechanisms, strict permission boundaries, and reliable tool calls. When you buy a packaged platform, you surrender control over context windows, retrieval strategies, and retry loops. That works fine for a basic internal Slack bot. It breaks down fast when your agent needs to touch core business logic.

Does that mean you should build your own foundation models? Absolutely not. Training an LLM or hosting base infrastructure from scratch is a massive waste of capital for ninety nine percent of companies. The smart move is a hybrid architecture. You buy raw model access through unified APIs, but you build the orchestration, memory architecture, and tool integrations yourself. If your internal team lacks the bandwidth to construct this middleware, working with external engineering partners who specialize in custom AI agent development is almost always cheaper than committing to a rigid SaaS subscription.

The real value of an agentic system lives in system integration. Your agent needs to talk to your legacy databases, your auth systems, and your internal APIs. A generic platform cannot anticipate how your company handles edge cases or data security. I always tell teams to build their own guardrails and state machines using standard code. TypeScript or Python handles deterministic flow control much better than a low code agent builder ever will. If you want real ROI, focus on custom workflow automation that tightly couples deterministic code with probabilistic LLM calls.

Vertical requirements make this build decision even clearer. Take regulated industries as an example. An off the shelf agent platform will fall flat on compliance standards, granular audit logs, and data retention rules. If you work on AI for Healthcare, every prompt, response, and intermediate tool call needs explicit logging and strict security boundaries. Buying a black box solution is a fast track to a security nightmare.

Here is the framework I use when making these decisions with engineering teams:

First, evaluate the control layer. If an agent failure breaks a critical customer flow or corrupts data, you must own the code that controls the agent execution logic. Never outsource your core control loop to a third party wrapper.

Second, look at the memory layer. Standard vector search is rarely enough for complex enterprise tasks. You need custom vector stores, key value caches, or relational tables tailored to your exact domain logic.

Third, treat prompts like source code. Store them in version control. Test them with automated CI/CD pipelines. Never hide them inside a vendor dashboard where engineers cannot run regression tests.

Stop looking for a magic software subscription that solves AI for your organization. The winners in this space will be engineering teams that treat AI agents as software architecture problems, not vendor procurement exercises. Build the orchestration, own the state, and plug in best of breed APIs for the raw intelligence layer.

If you are trying to figure out this balance for your own tech stack, take a look at Gaper. They help engineering leaders build custom AI infrastructure and scale specialized engineering teams without adding massive technical debt.