Agent fails after Composio tool execution despite valid cleaned response

[mayurjadhav2002]

When using a Composio Gmail tool with schema_modifier + after_execute sanitization, the tool executes successfully and returns a cleaned response, but the agent fails while generating the final reply, resulting in a generic error:

“Sorry — something went wrong while processing your request.”

This happens after tool execution, not during the tool call itself.

Basically GMAIL is fetching top 5 emails but with their complete html body including base64 which leads to input limit error.

How can fix this, any advice?

Environment
• Python
• Agents framework (Agent, Runner)
• Composio Gmail tools
• schema_modifier + after_execute
• Redis used for toolkit caching
• Model: gpt-4o-mini

MAX_TEXT_CHARS = 500          # absolute cap per field
MAX_SNIPPET_CHARS = 300       # snippet-specific cap

# Matches long base64 / binary-looking blobs
_BASE64_RE = re.compile(r"^[A-Za-z0-9+/=\s]{200,}$")

# Matches image URLs, cid refs, data URLs
_IMAGE_RE = re.compile(
    r"(data:image\/[a-zA-Z]+;base64,[^\s]+|cid:[^\s]+|https?:\/\/[^\s]+\.(png|jpg|jpeg|gif|webp))",
    re.IGNORECASE,
)


def _entropy(s: str) -> float:
    counts = Counter(s)
    total = len(s)
    return -sum((c / total) * math.log2(c / total) for c in counts.values())


def _looks_binary(s: str) -> bool:
    return bool(_BASE64_RE.match(s.strip())) or _entropy(s) > 4.5


def _sanitize_text(s: Any, *, max_len: int) -> str:
    if not isinstance(s, str):
        return ""

    # Remove image URLs / CID refs
    s = _IMAGE_RE.sub("[image removed]", s)

    # Drop binary / base64 junk
    if _looks_binary(s):
        return "[binary content removed]"

    # Hard clamp
    if len(s) > max_len:
        return s[:max_len] + "…"

    return s.strip()
# 
# Constants
# 

REDIS_TTL_SECONDS = 3600

GMAIL_LIGHT_FETCH_TOOLS = [
    "GMAIL_FETCH_EMAILS",
    "GMAIL_FETCH_MESSAGE_BY_MESSAGE_ID",
    "GMAIL_FETCH_MESSAGE_BY_THREAD_ID",
]


# 
# Gmail Schema Modifier (PRE-EXECUTION)
# Prevents heavy Gmail payloads before tool execution
# 

@schema_modifier(tools=GMAIL_LIGHT_FETCH_TOOLS)
def gmail_schema_modifier(tool: str, toolkit: str, schema: Tool) -> Tool:
    """
    Enforces lightweight Gmail fetches at schema level.
    This prevents payload / HTML / image blobs from ever being fetched.
    """
    props: Dict[str, Any] = schema.input_parameters.setdefault("properties", {})

    # Safe defaults (only if param exists)
    DEFAULTS = {
        "include_payload": False,
        "verbose": False,
        "ids_only": False,
        "max_results": 5,
    }

    for key, default in DEFAULTS.items():
        if key in props and isinstance(props[key], dict):
            props[key]["default"] = default

    # Hard-ban dangerous params
    for forbidden in ("page_token", "include_spam_trash"):
        props.pop(forbidden, None)

    return schema


# 
# Gmail Response Sanitizer (POST-EXECUTION)
# Ensures only text-safe, small data reaches the LLM
# 

def _safe_extract_headers(headers: Any) -> Dict[str, str]:
    if not isinstance(headers, list):
        return {}

    out: Dict[str, str] = {}
    for h in headers:
        name = h.get("name")
        value = h.get("value")
        if isinstance(name, str) and isinstance(value, str):
            out[name.lower()] = value
    return out


@after_execute(tools=GMAIL_LIGHT_FETCH_TOOLS)
def gmail_after_execute(
    tool: str,
    toolkit: str,
    response: ToolExecutionResponse,
) -> ToolExecutionResponse:
    try:
        if not response.get("successful"):
            return {
                "successful": False,
                "data": {},
                "error": response.get("error"),
            }

        data = response.get("data") or {}
        messages = data.get("messages")

        if not isinstance(messages, list):
            messages = []

        cleaned: List[Dict[str, str]] = []

        for msg in messages:
            payload = msg.get("payload") or {}
            headers = _safe_extract_headers(payload.get("headers"))

            cleaned.append({
                "from": _sanitize_text(headers.get("from"), max_len=200),
                "subject": _sanitize_text(headers.get("subject"), max_len=200),
                "date": _sanitize_text(headers.get("date"), max_len=64),
                "snippet": _sanitize_text(msg.get("snippet", ""), max_len=300),
            })

        print("cleaned response: ", cleaned)
        return cleaned

    except Exception as e:
        return {
            "successful": False,
            "data": {"messages": []},
            "error": str(e),
        }
# 
# Executor Agent Factory (Redis-safe, OAuth-safe)
# 

async def get_executor_agent(context: UserContext, prompt: str):
    """
    Returns a hardened Composio executor agent.
    Never throws. Never blocks. Never explodes token usage.
    """
    user_id = composio_user_id.format(user_id=context.user_id)
    redis_key = f"composio:toolkits:{user_id}"

    connected_toolkits: List[str] = []

    # ---------- Redis read (safe) ----------
    cached = await redis_client.get(redis_key)
    if cached:
        try:
            connected_toolkits = json.loads(cached)
            if not isinstance(connected_toolkits, list):
                connected_toolkits = []
        except Exception:
            await redis_client.delete(redis_key)

    # ---------- Fresh fetch if needed ----------
    if not connected_toolkits:
        try:
            accounts = composio.connected_accounts.list(user_ids=[user_id])
            connected_toolkits = [
                acc.toolkit.slug
                for acc in accounts.items
                if acc.toolkit and acc.toolkit.slug
            ]
            await redis_client.setex(
                redis_key,
                REDIS_TTL_SECONDS,
                json.dumps(connected_toolkits),
            )
        except Exception:
            connected_toolkits = []

    # ---------- Tool loading (safe) ----------
    tools = []
    if connected_toolkits:
        try:
            tools = composio.tools.get(
                user_id=user_id,
                toolkits=connected_toolkits,
                modifiers=[
                    gmail_schema_modifier,
                    gmail_after_execute,
                ],
            )
        except Exception:
            tools = []

    # ---------- Agent ----------
    agent = Agent(
        name="composio_agent",
        instructions=RECOMMENDED_PROMPT_PREFIX,
        model="gpt-4o-mini",
        tools=tools,
        model_settings=ModelSettings(include_usage=True),
    )

    return agent, tools

[KeepALifeUS]

Agent failing after tool execution with valid response is usually a parsing issue. Here is how to debug and fix:

Common Causes

1. Response format mismatch: Tool returns dict but agent expects string
2. Schema validation failure: Response missing required fields
3. Encoding issues: Special characters in response
4. Size limits: Response too large for context

Debugging Pattern

import json
from typing import Any

class ToolResponseDebugger:
    @staticmethod
    def validate_response(response: Any, expected_schema: dict = None) -> dict:
        issues = []
        
        # Check type
        if isinstance(response, str):
            try:
                parsed = json.loads(response)
                issues.append("Response is JSON string, might need parsing")
            except:
                pass
        
        # Check size
        response_str = str(response)
        if len(response_str) > 10000:
            issues.append(f"Response large: {len(response_str)} chars")
        
        # Check encoding
        try:
            response_str.encode("utf-8")
        except UnicodeError:
            issues.append("Encoding issues detected")
        
        # Check schema
        if expected_schema and isinstance(response, dict):
            for key in expected_schema.get("required", []):
                if key not in response:
                    issues.append(f"Missing required field: {key}")
        
        return {
            "valid": len(issues) == 0,
            "issues": issues,
            "response_type": type(response).__name__,
            "response_size": len(response_str)
        }

Fix: Response Normalizer

def normalize_tool_response(response: Any) -> str:
    """Ensure consistent response format for agents"""
    
    # Handle None
    if response is None:
        return json.dumps({"status": "success", "result": None})
    
    # Handle dict
    if isinstance(response, dict):
        # Truncate large values
        cleaned = {}
        for k, v in response.items():
            v_str = str(v)
            if len(v_str) > 5000:
                cleaned[k] = v_str[:5000] + "... (truncated)"
            else:
                cleaned[k] = v
        return json.dumps(cleaned, ensure_ascii=True, default=str)
    
    # Handle string
    if isinstance(response, str):
        if len(response) > 10000:
            return response[:10000] + "... (truncated)"
        return response
    
    # Fallback
    return json.dumps({"result": str(response)})

# Wrap Composio tool calls
async def safe_tool_call(tool, args):
    try:
        result = await tool.execute(**args)
        return normalize_tool_response(result)
    except Exception as e:
        return json.dumps({"error": str(e), "tool": tool.name})

Likely Fix for Your Case

Check if the response needs explicit JSON serialization before returning to the agent.

More on tool patterns: https://github.com/KeepALifeUS/autonomous-agents