The modern AGUI meets Agents Framework and MCP in this workshop, representing a real world scenario in Supply Chain Automation
In this post, we’re going to tackle a massive challenge in the agent space: Safety and Visibility. We are going to build a practical demo that connects two distinct MCP servers to a single agent service using the Microsoft Agents SDK and Azure OpenAI. To top it off, we’ll wrap it all in a lightweight web UI (AG-UI) that streams text, traces tool calls, and—crucially—gates state-changing actions behind human approval.
The Problem: Why Do We Need This?
As agent-based applications get more complex, we start hitting the same headaches over and over. We want agents to work with real backends, but we keep running into familiar pitfalls.
The “Black Box” Issue: Tool calls happen out of sight, so users have no idea what the agent is doing—instantly killing trust.
Tangled Logic: Backend logic gets crammed into prompts, turning into messy spaghetti that’s hard to test, deploy, and improve.
Unsafe Writes: An agent might update a database or delete a file without any human in the loop.
Our Goal: Keep backends modular with MCP, centralize the agent’s “brain,” and give users a UI that makes every tool action clear and trustworthy.
The Pitch: Backends stay as MCP tools, the agent brain lives in one service, and the UI makes tool activity fully transparent.
The Architecture
To solve this, we are using a microservices approach with Azure at its core.
High-Level Components
Policy MCP Server: Connects to Azure Blob Storage and serves as the source of truth for policy documents.
Order MCP Server: Connects to Azure SQL, managing structured order data.
Agent + AG-UI Service (FastAPI): The core of the system, linking to the MCP servers, running the agent through the Microsoft Agents SDK, and streaming events directly to the browser.
Web UI: A straightforward HTML/CSS/JavaScript frontend that displays the chat experience, tool traces, images, and human-approval cards.
The Data Flow (Mental Model)
Understanding the flow is key for effective debugging and observability. Here’s how a single request moves through the system:
Prompt: The browser sends a user prompt to the Agent Service.
Stream: The Agent Service instantly streams events back to the UI, including assistant text (token-by-token), tool-call traces (arguments and results), and custom UI elements like image cards or approval requests.
Execution: The Agent Service calls the appropriate MCP tools (Policy or Orders) via SSE and JSON-RPC.
Guardrails: For tools that change state (like updating an order), the agent pauses and explicitly requests human approval before proceeding.
Sample Client (AGUI) Code
# Convenience: if a tool returns an image URL (or JSON containing one), emit an AG-UI Custom event
# so clients can render it as a rich card.
def _looks_like_image_url(value: str) -> bool:
v = value.lower().split("?")[0].split("#")[0]
if not (v.startswith("http://") or v.startswith("https://")):
return False
return v.endswith((".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"))
image_url: str | None = None
if isinstance(result, str) and _looks_like_image_url(result.strip()):
image_url = result.strip()
else:
try:
parsed = json.loads(result)
if isinstance(parsed, dict):
for k in ("imageUrl", "image_url", "url", "image"):
v = parsed.get(k)
if isinstance(v, str) and _looks_like_image_url(v.strip()):
image_url = v.strip()
break
except Exception:
pass
if image_url:
emit({"type": "Custom", "name": "image", "value": {"url": image_url, "alt": tool_name}})
emit({"type": "StepFinished", "stepName": step_name})Sample Server (Policy Documents MCP Server)
MCP_server.call_tool()
async def call_tool(
name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
if name == "list_policy_documents":
try:
client = _blob_service_client()
storage = _safe_storage_info(client)
available = _list_blobs(limit=200)
return [
types.TextContent(
type="text",
text=json.dumps({"storage": storage, "available": available}, ensure_ascii=False),
)
]
except Exception as e:
return [types.TextContent(type="text", text=f"Error listing policies: {type(e).__name__}: {e}")]
if name == "read_policy_document":
requested = arguments.get("doc_name")
if not requested:
return [types.TextContent(type="text", text="Error: doc_name is required.")]
doc_name = _name_map(requested)
try:
client = _blob_service_client()
container = client.get_container_client(CONTAINER_NAME)
if not container.exists():
storage = _safe_storage_info(client)
return [
types.TextContent(
type="text",
text=(
"Policy container not found. "
+ json.dumps({"storage": storage}, ensure_ascii=False)
),
)
]
blob_client = container.get_blob_client(doc_name)
if not blob_client.exists():
storage = _safe_storage_info(client)
available = _list_blobs(limit=50)
return [
types.TextContent(
type="text",
text=(
f"Document '{doc_name}' not found in policy library. "
+ json.dumps({"storage": storage, "available": available}, ensure_ascii=False)
),
)
]
content = blob_client.download_blob().readall().decode("utf-8")
return [types.TextContent(type="text", text=content)]
except Exception as e:
return [types.TextContent(type="text", text=f"Error accessing policy library: {type(e).__name__}: {e}")]
raise ValueError(f"Unknown tool: {name}")Sample Server ( Orders MCP Server)
from mcp.server import Server
from mcp.server.sse import SseServerTransport
import mcp.types as types
import os
import pyodbc
import json
# Initialize MCP Server
mcp_server = Server("SQLOrderAgent")
# SQL Configuration
SQL_CONNECTION_STRING = os.getenv("SQL_CONNECTION_STRING")
def get_db_connection():
if not SQL_CONNECTION_STRING:
raise ValueError("SQL_CONNECTION_STRING environment variable is not set.")
return pyodbc.connect(SQL_CONNECTION_STRING)
def dict_from_row(cursor):
columns = [column[0] for column in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
def _column_exists(conn: pyodbc.Connection, table_name: str, column_name: str) -> bool:
cursor = conn.cursor()
cursor.execute(
"""
SELECT 1
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_NAME = ? AND COLUMN_NAME = ?
""",
(table_name, column_name),
)
return cursor.fetchone() is not None
@mcp_server.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="get_order_details",
description="Queries the SQL database for order details (status, priority, category, and optional fields like photo/address/remarks if present).",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "The ID of the order."}
},
"required": ["order_id"]
}
),
types.Tool(
name="get_order_address",
description="Returns an order's delivery address (and remarks if available) from the SQL database.",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "The ID of the order."}
},
"required": ["order_id"]
},
),Scenarios
Scenario 1: Image Rendering (Read-Only)
Prompt: “Show me the photo for order 5390”
What happens:
- The agent calls get_order_photo.
- The UI receives a Custom:image event.
- An image card is rendered directly inside the chat.
Scenario 2: Approval Gating (Human-in-the-Loop)
Prompt: “Set the photo for order 5390 to https://example.com/new_photo.jpg”
What happens:
- The agent detects a write operation.
- An Approval Card appears in the UI.
- Approve: Executes set_order_photo.
- Reject: Cancels the action entirely.
Scenario 3: Policy Lookup
Prompt: “List available policy docs, then read the hazardous policy.”
What happens: The agent queries the Policy MCP Server (Azure Blob Storage), lists the available files, and then reads the specific document you requested.
Lessons Learned & Design Patterns
Building this demo revealed some key insights for taking agentic systems to production.
MCP as a Boundary: MCP servers help keep domain tools cleanly separated, with clear ownership—Policy can run their own MCP server, and Orders can manage theirs independently.
Trust through Visibility: Streaming tool traces, including arguments and results, is crucial for smooth debugging and building genuine user trust.
First-Class Approval: Human approval works best when it’s a dedicated UI event that the frontend understands and enforces.
Operational Tips
Reuse Clients: Don’t recreate Azure SDK clients on every request—initialize them once at startup and reuse them.
Log Diagnostics: Always log tool latency and 429 errors to catch bottlenecks early.
Stable Schemas: Keep inputs and outputs small, explicit, and well-defined to cut down on hallucinations and unpredictable behavior.
This post explored a practical approach to building safe, observable agentic systems with MCP, Microsoft Agents SDK, and Azure-native services. By splitting domain logic into MCP servers, centralizing the agent’s “brain,” and streaming every tool action through a human-aware UI, we showed how to replace opaque, risky behavior with trust, control, and visibility—treating human approval as a true first-class interaction for powerful, responsible automation.
