Python
How to add pre- and post- processing to your Agents using Python.
Prerequisites
This tutorial assumes that you have set up MCP Toolbox with a basic agent as described in the local quickstart.
This guide demonstrates how to implement these patterns in your Toolbox applications.
Implementation
The following example demonstrates how to use ToolboxToolset with ADK’s pre and post processing hooks to implement pre and post processing for tool calls.
import asyncio
from datetime import datetime
from typing import Any, Dict, Optional
from copy import deepcopy
from google.adk import Agent
from google.adk.apps import App
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.tools.tool_context import ToolContext
from google.genai import types
from toolbox_adk import CredentialStrategy, ToolboxToolset, ToolboxTool
SYSTEM_PROMPT = """
You're a helpful hotel assistant. You handle hotel searching, booking and
cancellations. When the user searches for a hotel, mention it's name, id,
location and price tier. Always mention hotel ids while performing any
searches. This is very important for any operations. For any bookings or
cancellations, please provide the appropriate confirmation. Be sure to
update checkin or checkout dates if mentioned by the user.
Don't ask for confirmations from the user.
"""
# Pre processing
async def enfore_business_rules(
tool: ToolboxTool, args: Dict[str, Any], tool_context: ToolContext
) -> Optional[Dict[str, Any]]:
"""
Callback fired before a tool is executed.
Enforces business logic: Max stay duration is 14 days.
"""
tool_name = tool.name
print(f"POLICY CHECK: Intercepting '{tool_name}'")
if tool_name == "update-hotel" and "checkin_date" in args and "checkout_date" in args:
start = datetime.fromisoformat(args["checkin_date"])
end = datetime.fromisoformat(args["checkout_date"])
duration = (end - start).days
if duration > 14:
print("BLOCKED: Stay too long")
return {"result": "Error: Maximum stay duration is 14 days."}
return None
# Post processing
async def enrich_response(
tool: ToolboxTool,
args: Dict[str, Any],
tool_context: ToolContext,
tool_response: Any,
) -> Optional[Any]:
"""
Callback fired after a tool execution.
Enriches response for successful bookings.
"""
if isinstance(tool_response, dict):
result = tool_response.get("result", "")
elif isinstance(tool_response, str):
result = tool_response
else:
return None
tool_name = tool.name
if isinstance(result, str) and "Error" not in result:
if tool_name == "book-hotel":
loyalty_bonus = 500
enriched_result = f"Booking Confirmed!\n You earned {loyalty_bonus} Loyalty Points with this stay.\n\nSystem Details: {result}"
if isinstance(tool_response, dict):
modified_response = deepcopy(tool_response)
modified_response["result"] = enriched_result
return modified_response
else:
return enriched_result
return None
async def run_chat_turn(
runner: Runner, session_id: str, user_id: str, message_text: str
):
"""Executes a single chat turn and prints the interaction."""
print(f"\nUSER: '{message_text}'")
response_text = ""
async for event in runner.run_async(
user_id=user_id,
session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text=message_text)]),
):
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
response_text += part.text
print(f"AI: {response_text}")
async def main():
toolset = ToolboxToolset(
server_url="http://127.0.0.1:5000",
toolset_name="my-toolset",
credentials=CredentialStrategy.toolbox_identity(),
)
tools = await toolset.get_tools()
root_agent = Agent(
name="root_agent",
model="gemini-2.5-flash",
instruction=SYSTEM_PROMPT,
tools=tools,
# add any pre and post processing callbacks
before_tool_callback=enfore_business_rules,
after_tool_callback=enrich_response,
)
app = App(root_agent=root_agent, name="my_agent")
runner = Runner(app=app, session_service=InMemorySessionService())
session_id = "test-session"
user_id = "test-user"
await runner.session_service.create_session(
app_name=app.name, user_id=user_id, session_id=session_id
)
# First turn: Successful booking
await run_chat_turn(runner, session_id, user_id, "Book hotel with id 3.")
print("-" * 50)
# Second turn: Policy violation (stay > 14 days)
await run_chat_turn(
runner,
session_id,
user_id,
"Book a hotel with id 5 with checkin date 2025-01-18 and checkout date 2025-02-10",
)
await toolset.close()
if __name__ == "__main__":
asyncio.run(main())
You can also add model-level (before_model_callback, after_model_callback) and agent-level (before_agent_callback, after_agent_callback) hooks to intercept messages at different stages of the execution loop.
For more information, see the ADK Callbacks documentation.
The following example demonstrates how to use ToolboxClient with LangChain’s middleware to implement pre- and post- processing for tool calls.
import asyncio
from datetime import datetime
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import ToolMessage
from langchain_google_vertexai import ChatVertexAI
from toolbox_langchain import ToolboxClient
system_prompt = """
You're a helpful hotel assistant. You handle hotel searching, booking and
cancellations. When the user searches for a hotel, mention it's name, id,
location and price tier. Always mention hotel ids while performing any
searches. This is very important for any operations. For any bookings or
cancellations, please provide the appropriate confirmation. Be sure to
update checkin or checkout dates if mentioned by the user.
Don't ask for confirmations from the user.
"""
# Pre processing
@wrap_tool_call
async def enforce_business_rules(request, handler):
"""
Business Logic Validation:
Enforces max stay duration (e.g., max 14 days).
"""
tool_call = request.tool_call
name = tool_call["name"]
args = tool_call["args"]
print(f"POLICY CHECK: Intercepting '{name}'")
if name == "update-hotel":
if "checkin_date" in args and "checkout_date" in args:
try:
start = datetime.fromisoformat(args["checkin_date"])
end = datetime.fromisoformat(args["checkout_date"])
duration = (end - start).days
if duration > 14:
print("BLOCKED: Stay too long")
return ToolMessage(
content="Error: Maximum stay duration is 14 days.",
tool_call_id=tool_call["id"],
)
except ValueError:
pass # Ignore invalid date formats
# PRE: Code here runs BEFORE the tool execution
# EXEC: Execute the tool (or next middleware)
result = await handler(request)
# POST: Code here runs AFTER the tool execution
return result
# Post processing
@wrap_tool_call
async def enrich_response(request, handler):
"""
Post-Processing & Enrichment:
Adds loyalty points information to successful bookings.
Standardizes output format.
"""
# PRE: Code here runs BEFORE the tool execution
# EXEC: Execute the tool (or next middleware)
result = await handler(request)
# POST: Code here runs AFTER the tool execution
if isinstance(result, ToolMessage):
content = str(result.content)
tool_name = request.tool_call["name"]
if tool_name == "book-hotel" and "Error" not in content:
loyalty_bonus = 500
result.content = f"Booking Confirmed!\n You earned {loyalty_bonus} Loyalty Points with this stay.\n\nSystem Details: {content}"
return result
async def main():
async with ToolboxClient("http://127.0.0.1:5000") as client:
tools = await client.aload_toolset("my-toolset")
model = ChatVertexAI(model="gemini-2.5-flash")
agent = create_agent(
system_prompt=system_prompt,
model=model,
tools=tools,
# add any pre and post processing methods
middleware=[enforce_business_rules, enrich_response],
)
# Test post-processing
user_input = "Book hotel with id 3."
response = await agent.ainvoke(
{"messages": [{"role": "user", "content": user_input}]}
)
print("-" * 50)
last_ai_msg = response["messages"][-1].content
print(f"AI: {last_ai_msg}")
# Test Pre-processing
print("-" * 50)
user_input = "Update my hotel with id 3 with checkin date 2025-01-18 and checkout date 2025-02-20."
response = await agent.ainvoke(
{"messages": [{"role": "user", "content": user_input}]}
)
last_ai_msg = response["messages"][-1].content
print(f"AI: {last_ai_msg}")
if __name__ == "__main__":
asyncio.run(main())
You can also add model-level (wrap_model) and agent-level (before_agent, after_agent) hooks to intercept messages at different stages of the execution loop. See the LangChain Middleware documentation for details on these additional hook types.
Results
The output should look similar to the following.
Note
The exact responses may vary due to the non-deterministic nature of LLMs and differences between orchestration frameworks.
AI: Booking Confirmed! You earned 500 Loyalty Points with this stay.
AI: Error: Maximum stay duration is 14 days.