DEV Community: Darius Tran

Trace Every Step of Your AI Agent With a Small Logger

Darius Tran — Fri, 13 Nov 2026 02:53:42 +0000

When an agent gives a wrong answer, the first question is always the same: what did it actually do? Which tools did it call, in what order, and where did the time go?

If the only record is the chat transcript, you're guessing. A small trace logger fixes that, and you don't need a vendor to start. The code below is plain Node.js with no dependencies. It runs against a fake agent loop, so you can try it without an API key. I ran it on Node 24 and the assertions pass.

What a trace needs

Three things make a trace useful. Every step of one request shares an ID, so you can pull the whole story with one search. Every step records how long it took and whether it worked. And errors keep their message, because "something failed" isn't a clue.

I'd also log one line per step as JSON. Structured lines are easy to filter now and easy to ship to a real log system later.

Keep the trace ID visible to the user, too. When a customer writes in about a bad answer, a support person who can paste that ID into a search finds the whole run in seconds. Without it, they're matching timestamps by hand.

I work on AI Hive's agent platform, but this logger doesn't depend on it. It's the minimum I'd want in any agent before launch.

The logger

traced wraps any async step. It measures the time, records the outcome and rethrows the error, so wrapping a step never changes what your code does.

const { randomUUID } = require("node:crypto");

const events = [];

function logEvent(event) {
  events.push({ ts: new Date().toISOString(), ...event });
}

async function traced(traceId, step, fn) {
  const start = performance.now();
  try {
    const result = await fn();
    logEvent({ traceId, step, ms: Math.round(performance.now() - start), status: "ok" });
    return result;
  } catch (err) {
    logEvent({
      traceId,
      step,
      ms: Math.round(performance.now() - start),
      status: "error",
      error: String(err.message || err),
    });
    throw err;
  }
}

function slowestSteps(traceId, top = 3) {
  return events
    .filter((e) => e.traceId === traceId)
    .sort((a, b) => b.ms - a.ms)
    .slice(0, top)
    .map((e) => `${e.step}: ${e.ms} ms (${e.status})`);
}

In a real service you'd write each event to a log stream instead of an array. The array keeps the example short and lets the test inspect it.

Two details are worth copying. The error branch is deliberate: it logs and then rethrows, so tracing never hides a failure from the caller. And performance.now() is used for durations because it isn't affected by clock changes the way Date.now() can be.

A fake agent to trace

The agent below plans, looks up an order, checks a policy that times out, and answers. Each step is wrapped, and the failing policy check doesn't stop the run.

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function runAgent(question) {
  const traceId = randomUUID();
  await traced(traceId, "plan", () => sleep(30));
  const order = await traced(traceId, "tool:lookup_order", async () => {
    await sleep(120);
    return { id: "A1", status: "shipped" };
  });
  try {
    await traced(traceId, "tool:check_policy", async () => {
      await sleep(15);
      throw new Error("policy service timeout");
    });
  } catch (_) {
    // the agent carries on; the trace keeps the evidence
  }
  await traced(traceId, "final_answer", () => sleep(40));
  return { traceId, answer: `Order ${order.id} is ${order.status}.` };
}

Check that it works

This test runs the agent and confirms four steps were logged, one of them as an error, and that the slowest step is the order lookup.

const assert = require("node:assert");

(async () => {
  const { traceId, answer } = await runAgent("Where is my order?");
  assert.equal(answer, "Order A1 is shipped.");
  const mine = events.filter((e) => e.traceId === traceId);
  assert.equal(mine.length, 4);
  assert.equal(mine.filter((e) => e.status === "error").length, 1);
  assert.ok(slowestSteps(traceId, 1)[0].startsWith("tool:lookup_order"));
  console.log(slowestSteps(traceId));
})();

What to look for in real traces

Once real traffic flows through, a few patterns show up quickly. One tool that's slow on every request is usually a quick win. A step that errors but gets swallowed, like the policy check above, is a bug waiting to surface. And a trace with many repeated calls to the same tool often means the agent is stuck in a loop.

Add the user's role and the tool arguments to each event when your data rules allow it. The more of the decision you can see, the faster you can explain it.

Be careful with what you write down, though. Tool arguments and model outputs can contain personal data, so redact before logging and set a retention period. A trace store that keeps everything forever becomes its own risk.

You can also turn the slowest-steps helper into an alert. If the median time for one tool doubles overnight, you'll want to hear about it before the first complaint arrives.

Where to go next

When you run several agents that call each other, one trace ID per request isn't enough, and you'll want parent and child relationships between steps. I covered that, along with how to read the results, in tracing and debugging multi-agent systems.

What's the first thing you check when an agent misbehaves?

Year-End Checklist: Hardening Your AI Agents Before 2027

Darius Tran — Wed, 11 Nov 2026 01:42:25 +0000

The end of the year is a good moment to review an agent that shipped in the autumn and has now run for a few months. If it can read customer data or change records, check it yourself before an auditor does.

Here are ten checks I'd run, with small code samples for the ones that are easiest to get wrong. The code is plain Node.js with no dependencies. I ran every snippet below on Node 24, and the assertions at the end pass as written.

Access: who is the agent acting as?

Give the agent the permissions of the user who asked, not a shared admin account.
List every tool with the scopes it needs and a risk level.
Block anything irreversible until a named person approves it.

The first one is the check teams skip most often, because it takes work. A shared service account is convenient, and it means every prompt injection inherits the same reach. Passing the user's identity through is more effort, but it turns a possible data leak into a refused call.

This is the smallest version of that policy I could write. The check runs before the tool does, so the model never gets a chance to talk its way past it.

const policy = {
  lookup_order: { scopes: ["orders:read"], risk: "read" },
  issue_refund: { scopes: ["orders:write", "refunds:approve"], risk: "irreversible" },
};

const flags = { agentsEnabled: true };

function authorize(toolName, user, args = {}) {
  if (!flags.agentsEnabled) return { ok: false, reason: "kill_switch" };
  const rule = policy[toolName];
  if (!rule) return { ok: false, reason: "unknown_tool" };
  const missing = rule.scopes.filter((s) => !user.scopes.includes(s));
  if (missing.length > 0) return { ok: false, reason: "missing_scope", missing };
  if (rule.risk === "irreversible" && !args.approvedBy) {
    return { ok: false, reason: "needs_approval" };
  }
  return { ok: true };
}

When a call is refused, return the reason to the model as plain fields. A good model will ask the user for approval instead of retrying blindly.

If you want the full threat list behind these checks, the AI agent security guide for CISOs walks through it by category.

Data: what ends up in your logs?

Write down which fields the agent may read.
Redact obvious secrets and identifiers before anything is logged.
Set a retention period for prompts, tool results and logs, and delete on schedule.

Prompts and tool results often contain emails, tokens and card numbers. A few patterns catch the obvious cases before a line reaches your log store.

const PATTERNS = [
  [/[\w.+-]+@[\w-]+\.[\w.]+/g, "[email]"],
  [/\b(?:sk|key|tok)[-_][A-Za-z0-9_-]{8,}\b/g, "[secret]"],
  [/\b\d{13,19}\b/g, "[number]"],
];

function redact(text) {
  return PATTERNS.reduce((out, [re, label]) => out.replace(re, label), text);
}

Regexes catch the obvious cases and miss plenty of others, so treat this as a floor. It isn't a privacy program.

Logs are data too. If the redaction runs after the log line is written, it protects nothing, so apply it at the point where the event is created.

Actions: how do you stop it?

Keep a kill switch that refuses every tool call.
Rate limit each tool and each user.
Alert when the refusal rate or the error rate jumps.

The kill switch is the agentsEnabled flag in the first snippet. Set it to false and every call is refused. In production it should live in a config service or feature-flag system, so an operator can flip it without a deploy.

Rate limits matter more for agents than for people, because a looping agent can make thousands of calls before anyone looks. A ceiling per tool per minute turns that into a short, visible incident instead of a large bill.

Test every rule

Write one failing test per rule.

A rule without a test tends to disappear during the next refactor. These assertions cover the policy, the redaction and the kill switch.

const assert = require("node:assert");

const reader = { scopes: ["orders:read"] };
const approver = { scopes: ["orders:write", "refunds:approve"] };

assert.deepStrictEqual(authorize("lookup_order", reader), { ok: true });
assert.equal(authorize("issue_refund", reader).reason, "missing_scope");
assert.equal(authorize("issue_refund", approver).reason, "needs_approval");
assert.deepStrictEqual(authorize("issue_refund", approver, { approvedBy: "j.doe" }), { ok: true });
assert.equal(authorize("delete_everything", approver).reason, "unknown_tool");
assert.equal(
  redact("mail a.b@example.com key sk-abc12345xyz card 4111111111111111"),
  "mail [email] key [secret] card [number]"
);

flags.agentsEnabled = false;
assert.equal(authorize("lookup_order", reader).reason, "kill_switch");

Before the audit

Tick off the ten, keep the evidence, and name an owner. An auditor who finds a written policy, a log and a test for each rule tends to ask follow-up questions instead of raising findings. Keep that evidence somewhere they can reach without asking you: the policy file, a sample of logs and the results of the last test run.

I build AI agents for enterprise at AI Hive, and these ten checks come from the reviews I see there. Which one is missing from your list?

Building a Multi-Agent Supervisor in Python, Step by Step

Darius Tran — Mon, 09 Nov 2026 02:49:26 +0000

Most "multi-agent" demos are one agent with a longer prompt. A real supervisor does something simpler and more useful: it decides who should handle a request, passes the right context along, and stops when it should.

Quick disclosure: my day job is enterprise AI agent deployments at AI Hive. Nothing below depends on it. The code is plain Python with a stub where your model call goes.

Do you need more than one agent?

Probably not at first. Every extra agent adds latency, cost, and a new place for things to go wrong. I only split work when two jobs need different instructions, different tools, or different permissions.

A billing question and a product bug are a good example. They need different knowledge and different access, and mixing them in one prompt makes both answers worse. That's the case I'll build.

The pieces

There are three parts: a router that picks a specialist, the specialists themselves, and a loop that limits how many handoffs can happen. First, the model call and the specialists.

import json
from dataclasses import dataclass


def call_model(system: str, messages: list[dict]) -> str:
    """Replace with your provider's client. Must return plain text."""
    raise NotImplementedError


@dataclass
class Specialist:
    name: str
    system: str


SPECIALISTS = {
    "billing": Specialist(
        "billing",
        "You answer billing questions in three sentences or fewer. "
        "If the question is not about billing, reply with 'HANDOFF: ' "
        "and one line saying what is needed.",
    ),
    "tech": Specialist(
        "tech",
        "You troubleshoot product problems step by step. "
        "If the problem is really about billing, reply with 'HANDOFF: ' "
        "and one line saying what is needed.",
    ),
}

ROUTER_PROMPT = (
    "Route the customer message to exactly one specialist. "
    'Reply with JSON only: {"agent": "billing" or "tech" or "none", '
    '"reason": "short reason"}'
)

The HANDOFF: prefix is the whole handoff protocol. A specialist that realises the request isn't its job says so in a fixed format, and the supervisor reads it.

The router

The router asks the model for JSON and never trusts it blindly. If the output doesn't parse, or names an agent that doesn't exist, I treat it as "none".

def route(message: str) -> dict:
    raw = call_model(ROUTER_PROMPT, [{"role": "user", "content": message}])
    try:
        decision = json.loads(raw)
    except json.JSONDecodeError:
        return {"agent": "none", "reason": "router output was not valid JSON"}
    if not isinstance(decision, dict) or decision.get("agent") not in SPECIALISTS:
        return {"agent": "none", "reason": "no matching specialist"}
    return decision

That small check matters. Routers fail in boring ways, like extra text around the JSON, and a crash here takes the whole conversation with it.

Keep the router prompt short and list the allowed values explicitly. When you add a third specialist, update this prompt and your test messages first, before you touch the specialist itself.

The supervisor loop

Now the loop. It routes, calls the specialist, and checks for a handoff. It gives up after max_hops and returns an escalation instead of looping forever.

def handle(message: str, max_hops: int = 2) -> dict:
    log = []
    context = message
    for hop in range(max_hops):
        decision = route(context)
        log.append({"hop": hop, "route": decision})
        if decision["agent"] == "none":
            break
        specialist = SPECIALISTS[decision["agent"]]
        reply = call_model(specialist.system, [{"role": "user", "content": context}])
        log.append({"hop": hop, "agent": specialist.name, "reply": reply})
        if not reply.startswith("HANDOFF:"):
            return {"status": "answered", "reply": reply, "log": log}
        note = reply.removeprefix("HANDOFF:").strip()
        context = f"{message}\n\nNote from the previous specialist: {note}"
    return {"status": "escalated", "log": log}

Why is max_hops 2 and not 10? Every hop is another model call, so a bad handoff loop gets expensive fast. Two hops covers the usual case where the first specialist realises the request belongs elsewhere. Anything past that means the routing is wrong, and a person should see it. If you raise the limit, raise it for a reason you can point to in the logs.

Two details I'd keep in any version of this. The original message always travels with the handoff note, so the next specialist never works from a summary alone. And the function returns the full log, which is what you'll want open when a customer says the agent bounced them around.

Testing it without a model

Swap in a fake call_model and check the routing with plain asserts. No API key needed.

def fake_model(system: str, messages: list[dict]) -> str:
    text = messages[-1]["content"].lower()
    if system == ROUTER_PROMPT:
        if "invoice" in text:
            return '{"agent": "billing", "reason": "invoice question"}'
        if "crash" in text:
            return '{"agent": "tech", "reason": "product issue"}'
        return '{"agent": "none", "reason": "no match"}'
    return "Here is the answer."


call_model = fake_model

assert handle("I was charged twice on my invoice")["status"] == "answered"
assert handle("The app crashes on login")["status"] == "answered"
assert handle("What is the weather?")["status"] == "escalated"

Once this passes, collect 20 or 30 real messages with the specialist you expect for each one. Run them through the real router before you change a prompt, and again after. A routing change that looks harmless can quietly send one category to the wrong place.

What I'd add next

Three things, in this order. Per-specialist tool allow-lists, so billing can't call anything in the support system. A confidence field from the router, with a low score going straight to a human. And a trace ID on every log line so one conversation can be followed across agents.

If you'd rather build this kind of routing on a canvas, the Agent Studio and multi-agent sections of the AI Hive platform features page cover the visual and Python SDK sides of the same idea.

How do you decide when a second agent is worth the extra moving parts?

Claude Code in a Large Codebase: Review Workflows That Actually Help

Darius Tran — Fri, 06 Nov 2026 02:16:32 +0000

AI code review sounds great until it leaves forty comments on a pull request and thirty-eight of them are about naming. I've spent a while figuring out how to make Claude Code useful on a large codebase without that noise. Here's the workflow that stuck.

Give it the context a new teammate would need

The biggest quality jump came from writing down the things a senior engineer would tell a new hire. Claude Code reads a CLAUDE.md file in the repository, so that's where this goes.

# CLAUDE.md

## Review priorities
1. Correctness: logic errors, unhandled errors, race conditions.
2. Security: input validation, secrets, permission checks.
3. Breaking changes to public APIs or database schemas.

## Ignore in reviews
- Formatting and import order (handled by the linter).
- Naming preferences unless a name is actively misleading.

## Project facts
- All money values are integers in cents.
- Database access goes through the repository layer, never raw SQL in handlers.

This file does two jobs. It tells the model what to care about, and, just as important, what to skip. The "ignore" list cut most of the noise for me.

Review the diff, not the repo

On a large codebase, asking for a review of "the project" produces vague, generic feedback. Point it at the change instead. A simple pattern is to pipe the diff into a non-interactive run:

git diff main...HEAD | claude -p "Review this diff using the priorities in CLAUDE.md. \
List only issues that could cause a bug, a security problem, or a breaking change. \
For each, give the file, the line, and a one-sentence reason."

The explicit output format matters. Without it, you get essays. With it, you get a list a human reviewer can check in two minutes. Check the current Claude Code documentation for the exact flags in your version before wiring this into scripts.

Keep a human as the reviewer of record

I treat the AI review as a first pass, not an approval. A person still owns the merge decision. In practice, the AI catches the tedious things (a missing null check, an error swallowed in a catch block), and the human spends their attention on design and intent.

Two rules that helped the team trust the output:

Every AI comment has to point at a specific line. Comments without a location get ignored.
If an AI comment turns out to be wrong, someone adds a line to CLAUDE.md so it doesn't happen again.

That second rule turns false positives into documentation. After a few weeks, the file becomes a useful description of how the codebase actually works. I wrote up a longer look at real costs and production results from Claude Code reviews, including where it falls short.

Run security checks as a separate pass

I don't mix security review into the general review prompt. A single prompt that tries to check correctness, style, and security tends to do all three halfway. Instead, I run a second, narrow pass that only looks for things like secrets in code, missing permission checks, and unvalidated input reaching a database query.

The output goes into the same PR, labelled separately. Reviewers know that anything in the security list deserves a closer look, even when the general review looks clean.

Watch cost and scope

Large diffs cost more and get worse reviews. I ask for smaller pull requests anyway, and the AI review is one more reason. If a diff is huge, split the review by directory.

I also don't run it on every push. It runs when a PR is opened or marked ready, which keeps costs predictable.

Measure whether it's helping

The question isn't "is the AI smart?" It's "are we catching more real issues before merge, with less reviewer time?" Track how many AI comments lead to a code change, and how many get dismissed. If dismissals stay high, tighten the priorities in CLAUDE.md.

What's in your review context file, and which instruction cut the most noise for you?

I'm Darius Tran, and I work on getting AI into real engineering and operations workflows at AI Hive.

Prompt Versioning and Release Gates for LLM Apps

Darius Tran — Wed, 04 Nov 2026 03:20:03 +0000

The fastest way I know to break a working LLM feature is to "just tweak the prompt" on a Friday. Nothing crashes. The output simply gets a little worse, and nobody notices until a customer does.

So I've stopped treating prompts as config strings. They're code. They get versions, tests, and a gate in CI that can say no.

Put prompts in files, not in strings

The first step is boring and does most of the work. Every prompt lives in its own file, with a version and a short changelog note.

# prompts/ticket_summary.yaml
id: ticket_summary
version: 7
model: your-model-id
changelog: "Ask for the customer's requested outcome as a separate line."
template: |
  Summarize this support ticket in 3 bullet points.
  Then add one line: "Requested outcome: ..."
  Ticket:
  {ticket_text}

Now a prompt change shows up in a pull request, with a diff, a reviewer, and a version bump. That alone kills most silent regressions, because someone has to look.

Keep a small, honest eval set

Next, collect real cases. Not synthetic ones, at least not at first. I start with 30 to 50 examples pulled from production logs, each with what a good answer must contain.

# evals/ticket_summary_cases.py
CASES = [
    {
        "ticket_text": "My invoice for March was charged twice. Please refund one.",
        "must_include": ["refund", "Requested outcome"],
        "must_not_include": ["I don't know"],
    },
    # ... more real cases
]

String checks are crude, and that's fine for a gate. They catch the big breaks: missing sections, wrong format, refusals. For subtler quality, add a scored review later. I wrote about scoring beyond pass/fail in this piece on AI agent evaluation, which covers rubrics for actions as well as answers.

The release gate

The gate runs every case against the new prompt version and fails the build if the pass rate drops below the current version.

# evals/run_gate.py
import sys
import yaml
from evals.ticket_summary_cases import CASES

def call_model(prompt: str) -> str:
    """Wrap your LLM provider's SDK here."""
    raise NotImplementedError

def check(output: str, case: dict) -> bool:
    has_all = all(s.lower() in output.lower() for s in case["must_include"])
    has_none = not any(s.lower() in output.lower() for s in case["must_not_include"])
    return has_all and has_none

def pass_rate(template: str) -> float:
    passed = 0
    for case in CASES:
        output = call_model(template.format(ticket_text=case["ticket_text"]))
        passed += check(output, case)
    return passed / len(CASES)

if __name__ == "__main__":
    spec = yaml.safe_load(open("prompts/ticket_summary.yaml"))
    baseline = float(open("evals/baseline_ticket_summary.txt").read())
    score = pass_rate(spec["template"])
    print(f"v{spec['version']}: {score:.0%} (baseline {baseline:.0%})")
    sys.exit(0 if score >= baseline else 1)

Run it in CI on any change under prompts/. If the new version scores lower than the stored baseline, the merge is blocked. When it scores higher, update the baseline in the same PR.

A note on cost: running 50 cases on every commit adds up. I only trigger the gate on changes to the prompt folder or the model setting, not on every push.

Model upgrades go through the same gate

This is the part people skip. A new model version is a release too. Run the same eval set before switching, and keep the old model ID in config so rollback is a one-line change.

I've seen "better" models fail format checks that the old one passed every time. The gate catches that in minutes instead of a week of confused tickets.

Version the whole bundle, not just the prompt

A prompt rarely works alone. It depends on the model ID, the temperature, the tool definitions, and sometimes a system message that lives somewhere else. If you version only the template, you can still ship a "same prompt, different behaviour" change without noticing.

What I do now is treat these as one bundle. The YAML file holds the template, the model setting, and a hash of the tool schemas the prompt expects. The gate fails if any of them changes without a version bump. It sounds strict, but it's the difference between "the prompt didn't change" and "nothing changed", and only the second one is useful at 2am.

Practise the rollback

Rolling back is a one-line change only if someone has actually done it. Once a month, I revert the prompt to the previous version on a staging environment and run the gate. It takes ten minutes, and it proves the old version still passes against today's cases.

What this doesn't solve

String checks won't tell you whether a summary is actually useful. For that, you need a second layer: a sampled human review, or a scoring rubric run by a judge model, checked against human labels now and then.

Start with the gate anyway. It's an afternoon of work, and it turns "I think the prompt is fine" into a number.

How do you test prompt changes today, and has a model upgrade ever surprised you?

I'm Darius Tran, and I work on putting AI agents into production at AI Hive.

Wiring Claude Into Your Internal Tools With MCP: A Developer Walkthrough

Darius Tran — Mon, 02 Nov 2026 08:04:04 +0000

The Model Context Protocol (MCP) gives an AI client a standard way to call your tools. That's convenient, and it's also a new door into your systems. This walkthrough builds a small MCP server for an internal ticket database, and focuses on three habits I'd keep from the first line: narrow tools, limited data, and a log of every call.

The code is a sketch written from the MCP Python SDK documentation, not a tested library. The SDK changes quickly, so check the current docs for names and install steps before copying anything.

What we're building

An MCP server that lets an assistant look up support tickets and add an internal note. It has three tools:

search_tickets: read-only, returns a short list.
get_ticket: read-only, returns selected fields of one ticket.
add_internal_note: the only write, and it can only append a note.

Nothing else is exposed. No raw SQL, no delete, no export.

Step 1: Install and create the server

pip install "mcp[cli]"

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("tickets")

if __name__ == "__main__":
    mcp.run()  # stdio transport by default

FastMCP turns decorated Python functions into tools. It reads the type hints and docstring to describe each tool to the model, so write both carefully.

Step 2: Keep the database connection read-only

For the read tools, open the database in read-only mode. If a bug or a bad prompt tries to write, the connection refuses.

import sqlite3

def read_db():
    return sqlite3.connect("file:tickets.db?mode=ro", uri=True)

def write_db():
    return sqlite3.connect("tickets.db")

The write connection is only used by one tool, and that tool does one thing.

Step 3: Add narrow tools

Each tool returns only the fields the assistant needs. Customer emails and phone numbers stay out.

@mcp.tool()
def search_tickets(query: str, limit: int = 5) -> list[dict]:
    """Search ticket subjects. Returns id, subject and status for up to 10 tickets."""
    limit = min(limit, 10)
    with read_db() as db:
        rows = db.execute(
            "SELECT id, subject, status FROM tickets WHERE subject LIKE ? LIMIT ?",
            (f"%{query}%", limit),
        ).fetchall()
    return [{"id": r[0], "subject": r[1], "status": r[2]} for r in rows]


@mcp.tool()
def get_ticket(ticket_id: int) -> dict:
    """Get one ticket: subject, status, priority and the last public reply."""
    with read_db() as db:
        row = db.execute(
            "SELECT subject, status, priority, last_public_reply FROM tickets WHERE id = ?",
            (ticket_id,),
        ).fetchone()
    if row is None:
        return {"error": "ticket not found"}
    return dict(zip(["subject", "status", "priority", "last_public_reply"], row))


@mcp.tool()
def add_internal_note(ticket_id: int, note: str) -> dict:
    """Append an internal note to a ticket. Cannot edit or delete anything."""
    if len(note) > 1000:
        return {"error": "note too long"}
    with write_db() as db:
        db.execute(
            "INSERT INTO ticket_notes (ticket_id, note) VALUES (?, ?)",
            (ticket_id, note),
        )
    return {"ok": True}

Three details matter here. The limit is capped on the server, so the model can't ask for 10,000 rows. The queries use parameters, so input never becomes SQL. And the write tool appends to a separate notes table, so it can't rewrite history.

Step 4: Log every call

When someone asks what the assistant did last Tuesday, you want a real answer. A small decorator records each call as one JSON line.

import functools
import json
import logging
import time

logging.basicConfig(filename="mcp_calls.log", level=logging.INFO, format="%(message)s")

def audited(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.time()
        status = "ok"
        try:
            return fn(*args, **kwargs)
        except Exception as err:
            status = f"error: {type(err).__name__}"
            raise
        finally:
            logging.info(json.dumps({
                "tool": fn.__name__,
                "args": kwargs or list(args),
                "status": status,
                "ms": round((time.time() - start) * 1000),
            }))
    return wrapper

Apply it under the tool decorator, for example @mcp.tool() on top of @audited. Keep the order in mind: the tool decorator should receive the wrapped function, so the log runs on every call.

Step 5: Connect it to a client

For Claude Code, adding a local server is one command:

claude mcp add tickets -- python /path/to/server.py

Other clients use a config entry with a command and arguments. Check your client's docs for the exact file and format, because they differ.

Then test with real questions. Ask it to find a ticket, ask for something your tools don't support, and ask it to do something you didn't expose. The second and third tests show you where the boundary really is.

What I'd add before production

Authentication and per-user permissions. A local stdio server inherits your own access. A shared server needs real identity checks.
Rate limits. Cap calls per minute so a loop can't hammer the database.
Output review. Spot-check responses for data that shouldn't be there, like names inside free-text fields.
Approval for risky writes. Even an append-only tool deserves a human check if the note is customer-visible.
Tests. Keep a small set of prompts that must succeed and a set that must be refused, and run them after every change.

The principle

An MCP server is a contract about what an assistant can touch. The narrower the contract, the easier it is to review, test, and trust. Start with the smallest set of tools that does the job, and add more only when a real task needs them.

If you want the broader picture of connecting Claude to business systems, there's a guide on how to connect Claude to any tool.

What's the first internal tool you'd expose through MCP, and what's the one thing you'd refuse to expose?

I'm Darius Tran, an AI Implementation Specialist at AI Hive, working on getting AI agents into production for enterprise teams.

Claude's Context Window in Practice: Caching, Chunking, and Cost

Darius Tran — Thu, 29 Oct 2026 03:26:32 +0000

A bigger context window raises a design question that didn't exist a couple of years ago: if the model can read 100 pages at once, do you still need retrieval?

Sometimes you don't. Often you do. This post walks through three ways to give Claude your documents (stuff, cache, retrieve), with minimal Python for each and a rough way to compare cost. The code is a sketch written from the Anthropic documentation, not a tested library, so check parameter names and pricing against the current docs before you copy anything.

Start by counting tokens

Before choosing an approach, find out how big your material really is. The SDK can count tokens for a request without running it:

import anthropic

client = anthropic.Anthropic()

count = client.messages.count_tokens(
    model="claude-sonnet-5-5",
    system="You answer questions about our policy handbook.",
    messages=[{"role": "user", "content": handbook_text + "\n\nQuestion: ..."}],
)
print(count.input_tokens)

If the number is a small fraction of the model's context window, long context is on the table. If it's close to the limit, or your corpus is many times larger, retrieval is the only real option.

Option 1: put it all in the prompt, and cache it

For a stable document set that many requests share, such as a handbook, a product spec, or a contract template, you can send the whole thing each time and mark it for caching. Repeated requests then reuse the processed prefix instead of paying full price for it again.

response = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=500,
    system=[
        {"type": "text", "text": "You answer questions about our policy handbook."},
        {
            "type": "text",
            "text": handbook_text,
            "cache_control": {"type": "ephemeral"},
        },
    ],
    messages=[{"role": "user", "content": question}],
)

usage = response.usage
print(usage.cache_creation_input_tokens, usage.cache_read_input_tokens)

Three habits make caching work:

Put stable content first and changing content last. The cache matches on a prefix. A timestamp or a user name near the top defeats it.
Watch the usage fields. cache_read_input_tokens should be large on the second and later calls. If it stays at zero, something in your prefix is changing.
Know the limits. There's a minimum prefix length before caching applies, and cached entries expire after a short time unless they're used again. Both vary by model, so check the docs.

Option 2: chunk and retrieve

When the corpus is large, changes often, or has to respect per-user permissions, retrieval is the better fit. You split documents into chunks, find the few that matter for each question, and send only those.

A simple chunker that keeps paragraphs together and adds a small overlap:

def chunk(text, max_chars=3000, overlap=300):
    paragraphs = text.split("\n\n")
    chunks, current = [], ""
    for p in paragraphs:
        if len(current) + len(p) > max_chars and current:
            chunks.append(current)
            current = current[-overlap:]
        current += p + "\n\n"
    if current.strip():
        chunks.append(current)
    return chunks

Then retrieval and prompt assembly. embed here stands for whatever embedding model you use, since that's a separate choice:

import numpy as np

def top_chunks(question, chunks, chunk_vectors, k=5):
    q = np.array(embed(question))
    scores = [float(np.dot(q, v) / (np.linalg.norm(q) * np.linalg.norm(v)))
              for v in chunk_vectors]
    best = sorted(range(len(chunks)), key=lambda i: scores[i], reverse=True)[:k]
    return [chunks[i] for i in best]

def answer(question, chunks, chunk_vectors):
    context = "\n\n---\n\n".join(top_chunks(question, chunks, chunk_vectors))
    return client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=500,
        system="Answer only from the provided excerpts. If they don't contain the answer, say so.",
        messages=[{"role": "user", "content": f"Excerpts:\n{context}\n\nQuestion: {question}"}],
    )

Retrieval has its own failure modes: the right chunk may not rank in the top five, and answers that span several sections can get cut. Budget time to evaluate it, not just build it.

A rough cost comparison

Prices change, so I won't print any. This helper takes your own numbers. The write and read multipliers describe how cached input is billed relative to normal input, so take them from the current pricing page:

def monthly_input_cost(
    doc_tokens, calls_per_month, price_per_mtok,
    cache_write_mult=None, cache_read_mult=None,
):
    full = doc_tokens * calls_per_month / 1_000_000 * price_per_mtok
    if cache_write_mult is None:
        return full  # no caching
    # simplified: one cache write, the rest reads (ignores expiry gaps)
    write = doc_tokens / 1_000_000 * price_per_mtok * cache_write_mult
    reads = doc_tokens * (calls_per_month - 1) / 1_000_000 * price_per_mtok * cache_read_mult
    return write + reads

Run it for your own document size and question volume, then compare against the retrieval path, where each call sends only a few thousand tokens. Two cautions. Caching entries expire, so a low-traffic app will rewrite the cache more often than this simplified function assumes. And retrieval has costs beyond tokens: embeddings, a vector store, and the engineering time to keep the index fresh.

How I choose

Situation	I'd lean toward
Small, stable corpus, many questions	Long context with caching
Large or fast-changing corpus	Retrieval
Per-user access rules on documents	Retrieval, filtered by permission
Need exact quotes and citations	Retrieval, with chunk IDs
Few questions, one-off analysis	Plain long context, no caching

Whichever you pick, test with your own questions. Models can miss details buried deep inside very long inputs, and a small test set of real questions with known answers will tell you more than any benchmark.

What to log

Whatever the approach, record these per call: input tokens, cached tokens read and written, output tokens, latency, and which chunks or documents were included. When cost or quality moves, those fields tell you why.

I've gone deeper on window sizes and cost patterns in Claude context window limits and cost-smart patterns.

How big is the biggest document set you've tried to put in a single prompt, and did it hold up? I'd like to hear what worked.

I'm Darius Tran, an AI Implementation Specialist at AI Hive, working on getting AI agents into production for enterprise teams.

LangChain vs CrewAI vs AutoGen: What I'd Pick for an Enterprise Agent in 2026

Darius Tran — Tue, 27 Oct 2026 03:15:04 +0000

Every few weeks someone asks me which agent framework to use. The honest answer starts with a question: what does your agent need to survive in production? Features and GitHub stars matter less than state handling, observability, permissions, and testing.

This post compares LangChain (with LangGraph), CrewAI, and AutoGen on those four criteria. The code sketches are minimal and written from memory of each project's documentation. These libraries change quickly, so check the current docs and release notes before you copy anything.

The four criteria

State. Can you persist where a task is, resume it after a crash, and inspect it?
Observability. Can you trace every model call, tool call, and handoff for one task?
Permissions. Can you limit which tools an agent can use and require approval for risky actions?
Testing. Can you run a saved set of cases after every change and compare results?

A framework doesn't have to do all four itself. It has to make them possible without fighting it.

LangChain and LangGraph

LangGraph models an agent as a graph of steps that share a typed state. That fits criterion one well, because state is explicit and you can add persistence.

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

class State(TypedDict):
    question: str
    answer: str

def answer(state: State) -> dict:
    return {"answer": call_your_model(state["question"])}

graph = StateGraph(State)
graph.add_node("answer", answer)
graph.set_entry_point("answer")
graph.add_edge("answer", END)

app = graph.compile()
print(app.invoke({"question": "What is our refund window?"}))

What I like: explicit state, conditional edges, and a clear place to insert an approval step. What to watch: the ecosystem is large and moves fast, so pin versions and keep your own thin wrapper around it.

CrewAI

CrewAI describes agents by role and goal, then groups them into a crew that works through tasks. It's quick to start, and the role language is easy for non-engineers to read.

from crewai import Agent, Task, Crew

analyst = Agent(
    role="Support analyst",
    goal="Classify incoming tickets and draft a reply",
    backstory="You work on a customer support team.",
)

task = Task(
    description="Classify this ticket and draft a short reply: {ticket}",
    expected_output="A category and a draft reply",
    agent=analyst,
)

crew = Crew(agents=[analyst], tasks=[task])
result = crew.kickoff(inputs={"ticket": "My order arrived damaged."})
print(result)

What I like: fast prototyping and readable structure. What to watch: you'll need to add your own persistence, approval steps, and tracing for production, and role-based prompts can hide how decisions get made.

AutoGen

AutoGen focuses on conversations between agents. Recent versions split the project into layers, and the high-level chat API is async.

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    model = OpenAIChatCompletionClient(model="your-model-name")
    agent = AssistantAgent("helper", model_client=model)
    result = await agent.run(task="Summarize this ticket in one line.")
    print(result.messages[-1].content)

asyncio.run(main())

What I like: a clean model for multi-agent conversation and good building blocks. What to watch: check the project's current status and recommended successor before you commit, because the Microsoft agent tooling has been changing. Pick a framework whose maintainers' roadmap you've actually read.

Side by side

Criterion	LangGraph	CrewAI	AutoGen
State	Explicit, good fit	Add your own	Add your own
Observability	Tracing integrations	Add your own	Add your own
Permissions	Build approval nodes	Build around tools	Build around tools
Testing	Bring your own eval set	Bring your own eval set	Bring your own eval set

The last row is the important one. None of these gives you a test suite. You still need a saved set of cases and a script that runs them after every change.

What I'd pick

For a workflow with clear steps, approvals, and an audit trail, I'd start with LangGraph because explicit state makes the other criteria easier. For a quick prototype that business teams will read, CrewAI gets you there fastest. For research on multi-agent conversation, AutoGen is worth studying.

And if the real need is permissions, approvals, tracing, and model switching across several teams, you may not want to assemble those from a framework at all. At that point, compare a platform before you build the plumbing yourself.

A checklist before you commit

Can I persist and resume a task after a restart?
Can I see one trace per task with every call?
Can I require approval for specific tools?
Can I rerun fifty saved cases after each change?
Who maintains this in a year, and what is its roadmap?

If you want the longer comparison, I wrote up LangChain vs CrewAI vs AutoGen compared. I work on enterprise agent deployments at AI Hive. Which of the four criteria has caused you the most trouble so far?

Building a Webhook-Driven Agent: Async Patterns for Slow Enterprise APIs

Darius Tran — Mon, 26 Oct 2026 08:31:14 +0000

Most agent demos call an API and wait. That works until the API is an ERP that takes 40 seconds to answer, or a batch endpoint that takes ten minutes. Then the agent hangs, the request times out, a retry fires, and now you've created the same purchase order twice.

This post walks through the pattern I reach for instead: accept the task, hand the slow work to a queue, let the ERP call us back, and keep a polling job around for the days the callback never arrives. The code is Python with FastAPI and httpx. The ERP API is made up, so adapt the field names to yours.

Why synchronous waiting breaks

There are three problems with waiting inside the agent's request:

Timeouts. Load balancers and clients give up long before a slow system finishes.
Retries without protection. A client that retries a timed-out call can trigger the same side effect twice.
Wasted capacity. A worker blocked on one slow call can't serve anyone else.

The fix is to split "accept the task" from "finish the task".

The shape of the solution

The client sends a task with an Idempotency-Key header and gets 202 Accepted right away.
A background worker submits the job to the ERP, including our callback URL.
The ERP calls /webhooks/erp when it finishes, with an HMAC signature.
A reconciler polls the ERP for jobs that have been waiting too long, in case the webhook got lost.

Setup and state

For the demo, state lives in dictionaries. In production, use a database so a restart doesn't lose tasks.

import asyncio, hashlib, hmac, os, time, uuid
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Header, HTTPException, Request

ERP_URL = os.environ["ERP_URL"]
CALLBACK_URL = os.environ["CALLBACK_URL"]          # public URL of /webhooks/erp
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode()

tasks: dict[str, dict] = {}     # task_id -> state
by_key: dict[str, str] = {}     # idempotency key -> task_id
by_job: dict[str, str] = {}     # ERP job id -> task_id
queue: asyncio.Queue = asyncio.Queue()

Accepting tasks safely

The idempotency key is what makes retries harmless. If the same key arrives twice, we return the original task instead of creating a new one.

@asynccontextmanager
async def lifespan(app: FastAPI):
    jobs = [asyncio.create_task(worker()), asyncio.create_task(reconcile())]
    yield
    for job in jobs:
        job.cancel()

app = FastAPI(lifespan=lifespan)


@app.post("/tasks", status_code=202)
async def create_task(body: dict, idempotency_key: str = Header(...)):
    if idempotency_key in by_key:
        return tasks[by_key[idempotency_key]]

    task_id = str(uuid.uuid4())
    tasks[task_id] = {
        "id": task_id, "status": "queued", "input": body,
        "result": None, "attempts": 0, "created": time.time(),
    }
    by_key[idempotency_key] = task_id
    await queue.put(task_id)
    return tasks[task_id]


@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    if task_id not in tasks:
        raise HTTPException(404)
    return tasks[task_id]

The worker

The worker submits the job and records the ERP's job id. We pass our own task_id as a reference, so if the ERP supports idempotent submission, a retried submit won't create a duplicate.

async def worker():
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            task_id = await queue.get()
            task = tasks[task_id]
            try:
                resp = await client.post(f"{ERP_URL}/jobs", json={
                    "payload": task["input"],
                    "callback_url": CALLBACK_URL,
                    "reference": task_id,
                })
                resp.raise_for_status()
                job_id = resp.json()["job_id"]
                by_job[job_id] = task_id
                task.update(status="waiting_for_erp", job_id=job_id)
            except httpx.HTTPError:
                task["attempts"] += 1
                if task["attempts"] < 3:
                    await asyncio.sleep(2 ** task["attempts"])
                    await queue.put(task_id)
                else:
                    task["status"] = "failed"
            finally:
                queue.task_done()

The webhook

Two rules matter here. Verify the signature before you trust anything, and treat duplicate deliveries as normal. Webhooks are usually "at least once", so the same event can arrive twice.

@app.post("/webhooks/erp")
async def erp_webhook(request: Request, x_signature: str = Header(...)):
    raw = await request.body()
    expected = hmac.new(WEBHOOK_SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, x_signature):
        raise HTTPException(401, "bad signature")

    event = await request.json()
    task_id = by_job.get(event["job_id"])
    if task_id is None:
        return {"ok": True}                      # unknown job, acknowledge anyway

    task = tasks[task_id]
    if task["status"] in ("done", "failed"):
        return {"ok": True}                      # duplicate delivery, ignore

    ok = event["state"] == "succeeded"
    task.update(status="done" if ok else "failed", result=event.get("result"))
    return {"ok": True}

Return a 2xx quickly. If your agent needs to do more work with the result, enqueue that instead of doing it inside the handler.

The polling fallback

Callbacks get lost: a deploy, a firewall change, a vendor outage. A reconciler that checks old jobs turns "silently stuck forever" into "late by a few minutes".

async def reconcile():
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            await asyncio.sleep(60)
            for task in list(tasks.values()):
                stale = time.time() - task["created"] > 300
                if task["status"] != "waiting_for_erp" or not stale:
                    continue
                try:
                    resp = await client.get(f"{ERP_URL}/jobs/{task['job_id']}")
                    resp.raise_for_status()
                    job = resp.json()
                except httpx.HTTPError:
                    continue
                if job["state"] in ("succeeded", "failed"):
                    ok = job["state"] == "succeeded"
                    task.update(status="done" if ok else "failed", result=job.get("result"))

Webhook or polling?

I don't think it's either/or. Webhooks give you speed and cheap traffic. Polling gives you certainty. Using both costs about twenty extra lines and removes a whole category of 3 a.m. questions.

What this doesn't cover

A few things you'd still need before production: persistent storage, a dead-letter list for tasks that fail permanently, metrics on queue depth and callback delay, and a way to cancel a task. I'd also add an alert for any task stuck in waiting_for_erp for more than an hour.

If you're designing the API side of an agent and want to compare options beyond webhooks, I wrote up REST, WebSocket, and webhook patterns for agents. I work on enterprise agent deployments at AI Hive. How do you handle lost callbacks today: polling, retries from the sender, or something else?

A Cost-Aware Model Router for Claude in About 80 Lines of Python

Darius Tran — Wed, 21 Oct 2026 02:17:16 +0000

The fastest way to cut an LLM bill without hurting quality is to stop sending every request to your most capable model. Most traffic in a typical agent workflow is classification, extraction, or short replies. A smaller, faster model handles those fine. The bigger model should be reserved for the steps where it measurably changes the outcome.

This post walks through the simple router I use with Claude models. It's not a framework. It's a pattern you can adapt in an afternoon.

The idea

Every request gets a cheap first pass that decides how hard it is. Easy requests go to a fast model. Hard ones go to a stronger model. If the fast model's answer fails a check, the request is retried on the stronger one.

Three ingredients:

A difficulty signal: rules, a small classifier, or a short call to the fast model itself.
Routing rules kept in config, not hard-coded.
An escalation path when the cheaper answer isn't good enough.

The router

import anthropic
from dataclasses import dataclass

client = anthropic.Anthropic()

ROUTES = {
    "fast": {"model": "claude-sonnet-5", "max_tokens": 1024},
    "strong": {"model": "claude-opus-5-5", "max_tokens": 4096},
}

HARD_TASKS = {"multi_step_plan", "contract_review", "root_cause_analysis"}

@dataclass
class Request:
    task_type: str
    text: str
    needs_tools: bool = False

def pick_route(req: Request) -> str:
    if req.task_type in HARD_TASKS:
        return "strong"
    if len(req.text) > 20_000:
        return "strong"
    if req.needs_tools and req.task_type == "multi_step_plan":
        return "strong"
    return "fast"

def call(route: str, system: str, user: str):
    cfg = ROUTES[route]
    resp = client.messages.create(
        model=cfg["model"],
        max_tokens=cfg["max_tokens"],
        system=system,
        messages=[{"role": "user", "content": user}],
    )
    return resp, cfg["model"]

The rules here are deliberately boring. Task type and input size go a long way before you need anything clever. Check the current model names and pricing in Anthropic's documentation before copying this, since both change over time.

Escalation when the cheap answer isn't good enough

Routing up front catches the obvious cases. Escalation catches the rest:

import json

def validate(output_text: str, schema_keys: set) -> bool:
    try:
        data = json.loads(output_text)
    except json.JSONDecodeError:
        return False
    return schema_keys.issubset(data.keys())

def route_and_run(req: Request, system: str, schema_keys: set):
    route = pick_route(req)
    resp, model = call(route, system, req.text)
    text = resp.content[0].text
    if route == "fast" and not validate(text, schema_keys):
        resp, model = call("strong", system, req.text)
        text = resp.content[0].text
        log_escalation(req.task_type, reason="validation_failed")
    log_usage(req.task_type, model, resp.usage)
    return text

Validation can be anything cheap and deterministic: valid JSON, required fields present, a confidence field above a threshold, or a citation check for retrieval answers. Avoid using another LLM call as the validator for every request, or you lose most of the savings.

Measuring whether it works

A router without measurement is a guess. I log four things per request:

Task type
Model used
Input and output tokens
Whether it escalated, and why

From that, two numbers matter most:

Escalation rate per task type. If a task escalates more than about a third of the time, it probably belongs on the strong route from the start. Paying for two calls is worse than paying for one larger one.

Quality by route. Sample outputs from both routes each week and score them against the same rubric. The goal is equal quality at lower cost, not lower cost at any quality.

Where routing goes wrong

Routing on the wrong signal. Text length is a weak proxy for difficulty. A short question about a complex contract can be hard, and a long but simple form can be easy. Task type is usually the better signal.

Ignoring latency. The fast model is often fast as well as cheap. For user-facing steps, that matters as much as cost.

Forgetting tool use. Multi-step tool use with dependent calls is where larger models tend to be more reliable. I keep planning steps on the strong route even when the text is short.

Hard-coding models. Models change often. Keep routes in config so switching a model is a deploy-free change, and rerun your eval set when you do.

Adding a learned difficulty signal later

Rules get you surprisingly far, but eventually you'll want a better signal than task type. The cheapest upgrade I've found is to let your own logs train it.

After a few weeks of routing, you have labelled data for free: requests that the fast model handled and passed validation, and requests that escalated. A small classifier on simple features can learn which new requests are likely to escalate and send them straight to the strong route.

from sklearn.linear_model import LogisticRegression

def features(req: Request):
    return [
        len(req.text),
        req.text.count("?"),
        int(req.needs_tools),
        int(req.task_type in HARD_TASKS),
        count_entities(req.text),  # IDs, amounts, dates mentioned
    ]

X = [features(r) for r in logged_requests]
y = [r.escalated for r in logged_requests]
clf = LogisticRegression().fit(X, y)

def pick_route_learned(req: Request) -> str:
    p = clf.predict_proba([features(req)])[0][1]
    return "strong" if p > 0.4 else "fast"

Keep the threshold in config and tune it against your eval set. A lower threshold sends more traffic to the strong model, raising cost and lowering escalations. The right point depends on how much you value latency and how expensive a wasted fast call is for you.

Cost guardrails beyond routing

Routing reduces average cost. It doesn't stop a single runaway task. I add two simple guards:

A token budget per task, enforced in the agent loop. When a task crosses it, the agent summarizes its progress and hands off instead of continuing.
A daily budget per tenant or team, with an alert at 80 percent. This catches a new integration that suddenly sends ten times the expected traffic.

How much does it save?

It depends entirely on your traffic mix. If most of your requests are simple, routing makes a large difference. If most are complex reasoning, it helps less. The only reliable way to know is to log a week of real traffic by task type, then simulate the routing rules against it before turning them on.

For a deeper comparison of the two models, including where the larger one tends to justify its price, I wrote up when Sonnet is enough and when Opus earns its cost.

I work on enterprise agent deployments at AI Hive, where model routing is one of the first things we set up for cost control. How do you decide which requests get your most expensive model?

Building an MCP Server for Internal Tools: What I'd Do Differently Next Time

Darius Tran — Mon, 19 Oct 2026 08:10:56 +0000

The Model Context Protocol makes it easy to expose internal tools to an AI model. Easy enough that my first MCP server went from idea to working prototype in an afternoon. Getting it to a state where a security team would approve it took considerably longer.

This post covers what I'd do differently if I started again, with small code examples using the Python SDK.

A quick refresher

An MCP server exposes three kinds of things to a client such as Claude Desktop, an IDE, or your own agent:

Tools: functions the model can call, with a JSON schema for arguments.
Resources: read-only data the client can load into context.
Prompts: reusable prompt templates.

The client discovers what's available, and the model decides when to use it. The protocol handles the plumbing, which is exactly why it's tempting to expose everything.

Lesson 1: Fewer, narrower tools

My first server exposed a generic run_sql tool. It was flexible and impressive in demos. It was also impossible to secure, impossible to audit meaningfully, and the model regularly wrote queries that scanned entire tables.

The second version had specific tools instead:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("internal-ops")

@mcp.tool()
def get_open_tickets(team: str, limit: int = 20) -> list[dict]:
    """Return open support tickets for one team, newest first. Max 50."""
    limit = min(limit, 50)
    return tickets.find_open(team=team, limit=limit)

@mcp.tool()
def get_customer_summary(customer_id: str) -> dict:
    """Return plan, status, and last 5 invoices for one customer."""
    return crm.summary(customer_id)

Narrow tools are easier for the model to choose between, easier to permission, and easier to explain to an auditor. If a team needs flexible querying, that belongs in a reporting tool for humans, not in an agent's toolset.

Lesson 2: Identity has to flow through

The easiest setup runs the MCP server with one service account. Every request, from every user, has the same permissions. In a company, that's a non-starter.

What I do now is pass the end user's identity with each request and check it inside the tool:

@mcp.tool()
def get_customer_summary(customer_id: str, ctx: Context) -> dict:
    user = resolve_user(ctx)  # from the client's auth token
    if not policy.can_view_customer(user, customer_id):
        return {"error": "forbidden", "detail": "You don't have access to this customer."}
    audit.log(user=user.id, tool="get_customer_summary", customer_id=customer_id)
    return crm.summary(customer_id)

How you resolve the user depends on your transport and auth setup. The principle is the same either way: the tool enforces the user's rights, and the model never sees data the user couldn't see directly.

Lesson 3: Descriptions are part of the interface

The model reads your docstrings to decide which tool to use. Vague descriptions produce wrong tool choices, and those don't throw errors.

Things that helped:

Say what the tool returns and its limits ("Max 50", "last 5 invoices").
Say when not to use it ("Do not use for billing disputes; use get_invoice_detail").
Keep argument names self-explanatory. team beats t, and customer_id beats id.

I review tool descriptions in pull requests with the same care as API changes, because to the model, that's what they are.

Lesson 4: Return errors the model can act on

When something fails, return a structured result instead of raising an exception that becomes a generic error message.

return {"error": "not_found", "detail": "No customer with id CUST-123. Ask the user to confirm the ID."}

Good models use that detail to recover, usually by asking the user a clarifying question. A stack trace gives them nothing to work with.

Lesson 5: Treat resources as a data leak surface

Resources feel harmless because they're read-only. But a resource that returns "all open tickets" can put hundreds of customer records into the model's context in one go. Apply the same permission checks and size limits to resources as to tools, and prefer tools with filters when the data is sensitive.

Lesson 6: Log at the server, not only at the client

Client-side logs tell you what the model asked for. Server-side logs tell you what actually happened in your systems. For audits and incident reviews, the server log is the one that matters. I log user, tool, arguments (redacted), result size, status, and duration for every call.

Lesson 7: Version your server like an API

Once more than one client depends on your MCP server, changing a tool becomes a breaking change. Renaming an argument or tightening a return format can quietly degrade every agent that uses it, because models adapt to the old description in prompts and examples.

What helped:

Add new tools instead of changing existing ones in incompatible ways, then deprecate the old tool after clients move.
Keep a changelog of tool additions, description edits, and schema changes.
Run a small behavioural test suite against the server on every release: a few real requests per tool, checked for both correct calls and correct refusals.

def test_summary_requires_access():
    ctx = fake_context(user="analyst-no-crm")
    result = get_customer_summary("CUST-123", ctx)
    assert result["error"] == "forbidden"

Lesson 8: Decide who is allowed to run servers

On the client side, MCP makes it easy for anyone to connect a new server. That's great for developers and a headache for security. Within a few months, I've seen teams with a long tail of unofficial servers running on laptops, each with its own credentials.

A light process works: an approved list of servers, an owner for each, and a review before any server that touches production data is added. It doesn't need to be heavy. It needs to exist before the list gets long.

What I'd set up on day one next time

A short list of narrow tools, each with an owner.
User identity passed through and checked in every tool.
Docstrings reviewed like API documentation.
Structured, actionable error results.
Size limits on every tool and resource.
Server-side audit logging with redaction.

MCP is a great standard for connecting models to systems. It doesn't decide who should be allowed to do what, though. That part is still on us. If you want to see how a platform approaches connectors and permissions at scale, here's how AI Hive handles integrations.

I work on enterprise agent deployments at AI Hive. What's the first tool you exposed over MCP, and did you keep it?

Agreed since this step still requires lots of human effort to achieve

Darius Tran — Fri, 16 Oct 2026 03:10:07 +0000

Jonathan Gottfried for Daily Context

Oct 12

Agents Can’t Decide What’s Okay to Break

#hacktoberfest #security #aie #agents

3 min read