Agent Workflow
Learn how to implement the complete agent decision workflow using TraceMem's Agent MCP server.
Overview
Every agent decision follows a consistent pattern: create a decision envelope, read data, evaluate policies, handle outcomes, write data, and close the decision.
Workflow Pattern
- Create Decision Envelope: Wrap your decision in a decision envelope
- Read Data: Access data through Data Products with explicit purposes
- Evaluate Policy: Check if the action is allowed
- Handle Outcome:
- If
allow: Proceed with the action - If
deny: Rollback the decision - If
requires_exception: Request approval and poll for status
- If
- Write Data: If allowed/approved, write data through Data Products
- Close Decision: Commit or rollback the decision
Setting Up Your Client
Create a simple MCP client:
text
import requests
import json
class MCPClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Agent {api_key}",
"Content-Type": "application/json"
})
self.request_id = 0
def _next_id(self):
self.request_id += 1
return self.request_id
def _call(self, method, params=None):
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": method,
"params": params or {}
}
response = self.session.post(self.base_url, json=request)
response.raise_for_status()
result = response.json()
if "error" in result:
error = result["error"]
raise Exception(f"MCP Error {error['code']}: {error['message']}")
return result.get("result", {})
def initialize(self):
return self._call("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "my-agent", "version": "1.0.0"}
})
def call_tool(self, name, arguments):
result = self._call("tools/call", {
"name": name,
"arguments": arguments
})
# Extract text content from MCP response
if "content" in result and len(result["content"]) > 0:
text_content = result["content"][0].get("text", "{}")
try:
return json.loads(text_content)
except json.JSONDecodeError:
return {"raw_text": text_content}
return result
Core Workflow Implementation
Here's a complete example:
text
import os
import time
# Initialize client
mcp_url = os.getenv("MCP_AGENT_URL", "https://mcp.tracemem.com")
api_key = os.getenv("TRACEMEM_API_KEY")
client = MCPClient(mcp_url, api_key)
# Initialize session
client.initialize()
decision_id = None
try:
# 1. Create decision
decision = client.call_tool("decision_create", {
"intent": "customer.order.create",
"automation_mode": "propose",
"instance": "my-agent-v1"
})
decision_id = decision["decision_id"]
# 2. Read customer data
customer = client.call_tool("decision_read", {
"decision_id": decision_id,
"product": "customer_data", # Your data product ID
"purpose": "order_processing", # Must be in allowed_purposes
"query": {"customer_id": "123"}
})
# 3. Evaluate policy
policy_result = client.call_tool("decision_evaluate", {
"decision_id": decision_id,
"policy_id": "order_limit_v1", # Your policy ID
"inputs": {"order_amount": 1000.00}
})
outcome = policy_result["outcome"]
# 4. Handle policy outcome
if outcome == "allow":
# Write order
write_result = client.call_tool("decision_write", {
"decision_id": decision_id,
"product": "orders",
"purpose": "order_creation",
"mutation": {
"operation": "insert",
"records": [{
"customer_id": "123",
"total": 1000.00
}]
}
})
elif outcome == "requires_exception":
# Request approval
approval = client.call_tool("decision_request_approval", {
"decision_id": decision_id,
"title": "Order Approval",
"message": "Customer requesting order above policy limit...",
"require_rationale": True,
"expires_in_seconds": 3600
})
# Poll for approval status
max_wait = 300 # 5 minutes
waited = 0
while waited < max_wait:
time.sleep(5)
decision_status = client.call_tool("decision_get", {
"decision_id": decision_id
})
status = decision_status["status"]
if status == "approved":
# Write order
client.call_tool("decision_write", {
"decision_id": decision_id,
"product": "orders",
"purpose": "order_creation",
"mutation": {
"operation": "insert",
"records": [{
"customer_id": "123",
"total": 1000.00
}]
}
})
break
elif status == "rejected":
raise Exception("Approval rejected")
waited += 5
if waited >= max_wait:
raise Exception("Approval timeout")
elif outcome == "deny":
raise Exception("Action denied by policy")
# 5. Close decision
client.call_tool("decision_close", {
"decision_id": decision_id,
"action": "commit"
})
except Exception as e:
# On error, rollback
if decision_id:
client.call_tool("decision_close", {
"decision_id": decision_id,
"action": "rollback"
})
raise
Error Handling
Handle errors gracefully:
text
try:
result = client.call_tool("decision_read", {...})
except Exception as e:
if "401" in str(e) or "Unauthorized" in str(e):
print("Invalid API key")
elif "403" in str(e) or "Forbidden" in str(e):
print("Permission denied")
elif "429" in str(e):
print("Rate limited - retry later")
else:
print(f"Error: {e}")