fix: add MCP result validation, improve agent error handling and state management

- Add `_extract_mcp_content_text()` helper in naliia_tools.py to safely
  extract text from MCP results, preventing IndexError on empty content
- Replace all direct `result.content[0].text` accesses with safe helper
- Improve customer identification with preservation of existing state
- Add proper JSON parsing with fallback and error handling
- Simplify webhook state management (use agent's checkpointer internally)
- Update system prompt to remove check_mcp_connection tool reference
- Fix tests to use async mocks and correct expected values
This commit is contained in:
2026-03-19 21:11:28 -05:00
parent 12822784b0
commit 30345538f7
7 changed files with 231 additions and 91 deletions

View File

@@ -0,0 +1,139 @@
# Plan de Corrección: NaliiaTools - Puntos Críticos MCP
## Resumen
Este documento enumera las tareas de corrección identificadas en `src/naliiabot/bot/tools/naliia_tools.py`, priorizadas por gravedad.
---
## Tareas por Prioridad
### 🔴 PRIORIDAD ALTA
#### T-01: `create_schedule` retorna `True` sin validar el resultado del MCP
**Archivo:** `naliia_tools.py:283-285`
**Gravedad:** Crítica
**Descripción:** La función `schedule_appointment` siempre retorna `True` después de hacer la llamada MCP, ignorando completamente el resultado. Si el servidor falla o retorna un error, el agente asume que la cita fue creada exitosamente.
**Acción requerida:**
- [ ] Usar `_extract_mcp_content_text(result)` para extraer el contenido
- [ ] Validar que el resultado contenga un ID de cita válido
- [ ] Retornar `False` si la llamada falla o el resultado es inválido
- [ ] Loguear el contenido real del resultado (no solo `result`)
---
### 🔴 PRIORIDAD ALTA
#### T-02: `asyncio.run()` bloqueante en cada llamada MCP
**Archivo:** `naliia_tools.py:131, 158, 184, 213, 283`
**Gravedad:** Alta
**Descripción:** Cada llamada a herramienta invoca `asyncio.run()`, lo cual bloquea el event loop. En un servidor con múltiples requests, esto causa degradación secuencial.
**Acción requerida:**
- [ ] Opción A: Crear un `async_executor` con `ThreadPoolExecutor` para ejecutar las llamadas async
- [ ] Opción B: Refactorizar las herramientas para recibir el event loop ya corriendo
- [ ] Agregar timeouts a las llamadas (`asyncio.wait_for`)
---
### 🟡 PRIORIDAD MEDIA
#### T-03: Sin manejo de reconexión ni retry para `_MCP_CLIENT`
**Archivo:** `naliia_tools.py:15`
**Gravedad:** Media
**Descripción:** El cliente MCP es un singleton global sin retry automático. Si el servidor se cae, todas las herramientas fallan hasta reiniciar.
**Acción requerida:**
- [ ] Crear wrapper con retry exponencial (3 intentos, backoff)
- [ ] Implementar health check periódico
- [ ] Agregar timeout global (sugerido: 30s)
---
### 🟡 PRIORIDAD MEDIA
#### T-04: `register_customer` tipo de retorno incorrecto
**Archivo:** `naliia_tools.py:106`
**Gravedad:** Media
**Descripción:** La función declara `-> bool` pero retorna `_extract_mcp_content_text(result)` que es `str`.
**Acción requerida:**
- [ ] Cambiar tipo de retorno a `-> str`
- [ ] Actualizar docstring para reflejar que retorna el ID del cliente como string
---
### 🟡 PRIORIDAD MEDIA
#### T-05: `check_mcp_connection` no se usa y tiene lógica incorrecta
**Archivo:** `naliia_tools.py:28-44`
**Gravedad:** Media
**Descripción:** La función no está incluida en `get_tools()`. Además, `return True` está hardcodeado dentro de `call_tool()` ignorando el resultado real de `ping()`.
**Acción requerida:**
- [ ] Opción A: Conectar la función al sistema de health checks
- [ ] Opción B: Eliminar la función si no tiene uso previsto
- [ ] Si se mantiene, corregir la lógica para retornar el estado real del ping
---
### 🟢 PRIORIDAD BAJA
#### T-06: Sin timeout en llamadas MCP
**Archivo:** `naliia_tools.py` (todas las llamadas async)
**Gravedad:** Baja
**Descripción:** Si el servidor MCP no responde, las llamadas bloquean indefinidamente.
**Acción requerida:**
- [ ] Envolver todas las llamadas con `asyncio.wait_for(coro, timeout=30)`
- [ ] Capturar `asyncio.TimeoutError` y retornar error apropiado
---
### 🟢 PRIORIDAD BAJA
#### T-07: Mejorar logging de resultados MCP
**Archivo:** `naliia_tools.py:279`
**Gravedad:** Baja
**Descripción:** `logger.info(result)` loguea el objeto completo en lugar del contenido útil.
**Acción requerida:**
- [ ] Cambiar a `logger.info(_extract_mcp_content_text(result))`
- [ ] Agregar logging de errores con nivel `ERROR`
---
## Dependencias entre Tareas
```
T-02 (asyncio.run) ──┬── Requerido por T-06 (timeout)
└── Contexto para T-03 (retry wrapper)
T-03 (retry wrapper) ── Requerido por T-05 (health check)
T-01 (create_schedule) ── Independiente
T-04 (register_customer) ── Independiente
T-07 (logging) ── Independiente
```
---
## Orden de Implementación Sugerido
1. **T-01** - Fix crítico: `create_schedule` retornando `True` siempre
2. **T-02** - Mejora de rendimiento: reemplazar `asyncio.run()`
3. **T-06** - Complemento de T-02: agregar timeouts
4. **T-03** - Resiliencia: retry y reconexión
5. **T-04** - Fix de tipos: corregir retorno de `register_customer`
6. **T-05** - Decisión: conectar o eliminar `check_mcp_connection`
7. **T-07** - Mejora: logging consistente
---
## Métricas de Éxito
- [ ] Todas las llamadas MCP tienen timeout configurado
- [ ] `schedule_appointment` retorna `False` cuando falla el MCP
- [ ] No hay `asyncio.run()` en funciones llamadas desde el agent
- [ ] Tipos de retorno corresponden a los valores retornados

View File

@@ -66,11 +66,6 @@ class Agent:
"""Retorna configuración actual.""" """Retorna configuración actual."""
return self._config return self._config
@property
def checkpointer(self):
"""Retorna el checkpointer para acceso al estado."""
return self._checkpointer
@property @property
def graph(self) -> StateGraph: def graph(self) -> StateGraph:
"""Expone el grafo para inspección en tests.""" """Expone el grafo para inspección en tests."""
@@ -139,7 +134,7 @@ class Agent:
agent_builder.add_node("llm_call", self._llm_call_node) agent_builder.add_node("llm_call", self._llm_call_node)
agent_builder.add_node("tool_node", self._tool_node) agent_builder.add_node("tool_node", self._tool_node)
agent_builder.add_edge(START, "llm_call") agent_builder.add_edge(START, "identify_customer")
agent_builder.add_edge("identify_customer", "llm_call") agent_builder.add_edge("identify_customer", "llm_call")
agent_builder.add_conditional_edges( agent_builder.add_conditional_edges(
"llm_call", self._should_continue, {"tool_node": "tool_node", END: END} "llm_call", self._should_continue, {"tool_node": "tool_node", END: END}
@@ -151,10 +146,14 @@ class Agent:
def _identify_customer_node(self, state: MessagesState) -> dict: def _identify_customer_node(self, state: MessagesState) -> dict:
""" """
Nodo de identificación del cliente. Nodo de identificación del cliente.
Busca el cliente por teléfono y actualiza el estado. Extrae customer_phone del estado restaurado y actualiza customer_name, customer_id.
Si ya está identificado, preserva los valores existentes.
""" """
from ..logger import logger from ..logger import logger
if state.get("customer_name") and state.get("customer_id", 0) > 0:
return {}
customer_phone = state.get("customer_phone", "") customer_phone = state.get("customer_phone", "")
if not customer_phone: if not customer_phone:
return {"customer_name": "Cliente Anónimo", "customer_id": 0} return {"customer_name": "Cliente Anónimo", "customer_id": 0}
@@ -166,10 +165,19 @@ class Agent:
try: try:
result = find_customer_tool.invoke({"identifier": customer_phone}) result = find_customer_tool.invoke({"identifier": customer_phone})
customer_data = json.loads(result) if result else {}
if not result:
customer_data = {}
else:
try:
customer_data = (
json.loads(result) if isinstance(result, str) else result
)
except json.JSONDecodeError:
customer_data = {}
if isinstance(customer_data, list): if isinstance(customer_data, list):
customer_data = customer_data[0] if customer_data else {} customer_data = customer_data[0] if len(customer_data) > 0 else {}
if customer_data and customer_data.get("party.", {}).get("id", 0): if customer_data and customer_data.get("party.", {}).get("id", 0):
customer_name = customer_data.get("party.", {}).get( customer_name = customer_data.get("party.", {}).get(
@@ -236,9 +244,23 @@ class Agent:
if tool_call.get("name") == "find_customer_by_identifier": if tool_call.get("name") == "find_customer_by_identifier":
try: try:
customer_data = json.loads(result.content) if result.content else {} if not result.content:
customer_data = {}
else:
try:
customer_data = (
json.loads(result.content)
if isinstance(result.content, str)
else result.content
)
except json.JSONDecodeError:
customer_data = {}
if isinstance(customer_data, list): if isinstance(customer_data, list):
customer_data = customer_data[0] if customer_data else {} customer_data = (
customer_data[0] if len(customer_data) > 0 else {}
)
if customer_data and customer_data.get("party.", {}).get("id", 0): if customer_data and customer_data.get("party.", {}).get("id", 0):
updates["customer_name"] = customer_data.get("party.", {}).get( updates["customer_name"] = customer_data.get("party.", {}).get(
"name", "Cliente" "name", "Cliente"
@@ -246,7 +268,7 @@ class Agent:
updates["customer_id"] = customer_data.get("party.", {}).get( updates["customer_id"] = customer_data.get("party.", {}).get(
"id", 0 "id", 0
) )
except (json.JSONDecodeError, KeyError): except (json.JSONDecodeError, KeyError, IndexError, TypeError):
pass pass
if self._post_tool_hook: if self._post_tool_hook:

View File

@@ -5,8 +5,6 @@ prompts:
Preséntate como Naliia. Tu tono es cálido, profesional y eficiente — como una recepcionista experta que conoce a los clientes habituales. Preséntate como Naliia. Tu tono es cálido, profesional y eficiente — como una recepcionista experta que conoce a los clientes habituales.
Si el cliente está identificado en el estado de la conversación, dirígete a él por su nombre. Después de saludarlo por primera vez con su nombre, continúa usándolo durante la conversación.
</Identidad> </Identidad>
<Reglas de Conversación> <Reglas de Conversación>
@@ -20,10 +18,10 @@ prompts:
IMPORTANTE - Identificación del Cliente: IMPORTANTE - Identificación del Cliente:
- Cuando el usuario proporcione un número de teléfono, documento de identidad o email, - Cuando el usuario proporcione un número de teléfono, documento de identidad o email,
DEBES inmediatamente llamar a la herramienta `find_customer_by_identifier` con ese valor. DEBES llamar a la herramienta `find_customer_by_identifier` con ese valor.
- No asumas que ya conoces al cliente, verifica su identidad usando la herramienta. Ejemplo: Si el usuario dice "mi teléfono es 3016859278" o "3016859278", llama inmediatamente
- Ejemplo: Si el usuario dice "mi teléfono es 3016859278" o "3016859278", llama inmediatamente
a `find_customer_by_identifier` con ese identificador. a `find_customer_by_identifier` con ese identificador.
- No asumas que ya conoces al cliente, verifica su identidad usando la herramienta.
- Si el cliente no se encuentra en el sistema, indícale amablemente que no aparece registrado - Si el cliente no se encuentra en el sistema, indícale amablemente que no aparece registrado
y pídele verificar sus datos. y pídele verificar sus datos.
@@ -61,6 +59,5 @@ prompts:
- find_service_centers: Consultar los centros de atención o servicio disponibles. - find_service_centers: Consultar los centros de atención o servicio disponibles.
- find_products_and_services: Consultar los productos y servicios disponibles con precios. - find_products_and_services: Consultar los productos y servicios disponibles con precios.
- find_customer_by_identifier: Buscar cliente por teléfono, email o número de documento. SIEMPRE llama esta herramienta cuando el usuario proporcione un identificador. - find_customer_by_identifier: Buscar cliente por teléfono, email o número de documento. SIEMPRE llama esta herramienta cuando el usuario proporcione un identificador.
- check_mcp_connection: Verifica la conexión con el Servidor MCP, NO dar información al cliente.
</Tools Disponibles Para Brindar Atención al Cliente> </Tools Disponibles Para Brindar Atención al Cliente>
""" """

View File

@@ -15,6 +15,35 @@ MCP_SERVER_URL = "http://192.168.58.109:3001/mcp"
_MCP_CLIENT = Client(MCP_SERVER_URL) _MCP_CLIENT = Client(MCP_SERVER_URL)
def _extract_mcp_content_text(result, fallback: str = "") -> str:
if not result.content:
logger.warning("MCP result content is empty")
return fallback
if not hasattr(result.content[0], "text"):
logger.warning("MCP result content[0] has no 'text' attribute")
return fallback
return result.content[0].text
def check_mcp_connection(self) -> str:
"""
Verifica la conexión con el servidor MCP.
Returns:
str: "connected" si el servidor está disponible, "disconnected" si no hay conexión.
"""
async def call_tool():
async with _MCP_CLIENT:
result = await _MCP_CLIENT.ping()
logger.info(f"MCP connection status: {result}")
return True
is_connected = asyncio.run(call_tool())
return "connected" if is_connected else "disconnected"
class NaliiaTools: class NaliiaTools:
def __init__(self): def __init__(self):
self._session_active = False self._session_active = False
@@ -29,27 +58,8 @@ class NaliiaTools:
self.find_service_centers, self.find_service_centers,
self.find_products_and_services, self.find_products_and_services,
self.find_customer_by_identifier, self.find_customer_by_identifier,
self.check_mcp_connection,
] ]
@tool
def check_mcp_connection(self) -> str:
"""
Verifica la conexión con el servidor MCP.
Returns:
str: "connected" si el servidor está disponible, "disconnected" si no hay conexión.
"""
async def call_tool():
async with _MCP_CLIENT:
result = await _MCP_CLIENT.ping()
logger.info(f"MCP connection status: {result}")
return True
is_connected = asyncio.run(call_tool())
return "connected" if is_connected else "disconnected"
@tool @tool
def get_tomorrow_date() -> str: def get_tomorrow_date() -> str:
""" """
@@ -64,7 +74,7 @@ class NaliiaTools:
- Utiliza la zona horaria de Colombia (America/Bogota, UTC-5). - Utiliza la zona horaria de Colombia (America/Bogota, UTC-5).
""" """
td = timedelta(days=1) td = timedelta(days=1)
tomorrow = datetime.now().date() + td tomorrow = datetime.now(tz=ZoneInfo(TIMEZONE)).date() + td
tomorrow_date_formated = tomorrow.isoformat() tomorrow_date_formated = tomorrow.isoformat()
logger.info(tomorrow_date_formated) logger.info(tomorrow_date_formated)
@@ -106,26 +116,21 @@ class NaliiaTools:
Ejemplo: [1] Ejemplo: [1]
""" """
identifiers = { identifiers = {"type": "mobile", "code": cellphone}
"type": "mobile",
"code": cellphone
}
async def call_tool(): async def call_tool():
async with _MCP_CLIENT: async with _MCP_CLIENT:
result = await _MCP_CLIENT.call_tool( result = await _MCP_CLIENT.call_tool(
'create_customer', { "create_customer", {"name": name, "identifiers": identifiers}
"name": name, )
"identifiers": identifiers
})
logger.info(result.content[0].text) logger.info(_extract_mcp_content_text(result))
return result return result
result = asyncio.run(call_tool()) result = asyncio.run(call_tool())
return result.content[0].text return _extract_mcp_content_text(result)
@tool @tool
def find_customer_by_identifier(identifier: str) -> str: def find_customer_by_identifier(identifier: str) -> str:
@@ -147,12 +152,12 @@ class NaliiaTools:
result = await _MCP_CLIENT.call_tool( result = await _MCP_CLIENT.call_tool(
"find_customer_by_identifier", {"identifier": identifier} "find_customer_by_identifier", {"identifier": identifier}
) )
logger.info(result.content[0].text) logger.info(_extract_mcp_content_text(result))
return result return result
result = asyncio.run(call_tool()) result = asyncio.run(call_tool())
return result.content[0].text return _extract_mcp_content_text(result)
@tool @tool
def find_service_centers(): def find_service_centers():
@@ -172,13 +177,13 @@ class NaliiaTools:
async def call_tool(): async def call_tool():
async with _MCP_CLIENT: async with _MCP_CLIENT:
result = await _MCP_CLIENT.call_tool("find_service_centers", None) result = await _MCP_CLIENT.call_tool("find_service_centers", None)
logger.info(result.content[0].text) logger.info(_extract_mcp_content_text(result))
return result return result
result = asyncio.run(call_tool()) result = asyncio.run(call_tool())
return result.content[0].text return _extract_mcp_content_text(result)
@tool @tool
def find_products_and_services(): def find_products_and_services():
@@ -201,13 +206,13 @@ class NaliiaTools:
async def call_tool(): async def call_tool():
async with _MCP_CLIENT: async with _MCP_CLIENT:
result = await _MCP_CLIENT.call_tool("find_products_and_services", None) result = await _MCP_CLIENT.call_tool("find_products_and_services", None)
logger.info(result.content[0].text) logger.info(_extract_mcp_content_text(result))
return result return result
result = asyncio.run(call_tool()) result = asyncio.run(call_tool())
return result.content[0].text return _extract_mcp_content_text(result)
@tool(args_schema=ScheduleSchema) @tool(args_schema=ScheduleSchema)
def schedule_appointment( def schedule_appointment(

View File

@@ -65,28 +65,10 @@ async def webhook_chat(request: Request, agent=Depends(get_agent)):
config = {"configurable": {"thread_id": thread_id}} config = {"configurable": {"thread_id": thread_id}}
existing_state = await agent.checkpointer.aget(config) initial_state = {
"messages": [HumanMessage(content=user_message)],
if existing_state is None: "customer_phone": customer_phone,
initial_state = { }
"messages": [HumanMessage(content=user_message)],
"llm_calls": 0,
"customer_phone": customer_phone,
"customer_name": "",
"customer_id": 0,
}
logger.info(f"Nueva conversación iniciada con cliente: {customer_phone}")
else:
channel_values = existing_state.get("channel_values", existing_state)
initial_state = {
"messages": channel_values.get("messages", [])
+ [HumanMessage(content=user_message)],
"llm_calls": channel_values.get("llm_calls", 0),
"customer_phone": customer_phone,
"customer_name": channel_values.get("customer_name", ""),
"customer_id": channel_values.get("customer_id", 0),
}
logger.info(f"Conversación reanudada con cliente: {customer_phone}")
agent_response = await agent.ainvoke(initial_state, config=config) agent_response = await agent.ainvoke(initial_state, config=config)
agent_response_content = agent_response["messages"][-1].content agent_response_content = agent_response["messages"][-1].content

View File

@@ -32,7 +32,7 @@ class TestAgentConstruction:
agent = Agent(model=mock_model) agent = Agent(model=mock_model)
assert agent.config.max_iterations == 200 assert agent.config.max_iterations == 10
assert "helpful assistant" in agent.config.system_prompt assert "helpful assistant" in agent.config.system_prompt
assert agent.tools == [] assert agent.tools == []

View File

@@ -1,4 +1,5 @@
import pytest import pytest
from unittest.mock import AsyncMock
from types import SimpleNamespace from types import SimpleNamespace
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from naliiabotapi.main import app from naliiabotapi.main import app
@@ -12,27 +13,21 @@ class TestWebhookChatBot:
async def tests_webhook_success(self, send_message_payload): async def tests_webhook_success(self, send_message_payload):
"""Test that the webhook endpoint returns a successful response. """Test that the webhook endpoint returns a successful response.
Se crea un `mock_agent` con `invoke` mockeado y se inyecta Se crea un `mock_agent` con `ainvoke` mockeado y se inyecta
mediante `app.dependency_overrides` antes de realizar la petición. mediante `app.dependency_overrides` antes de realizar la petición.
""" """
# Crear mock del agente y su invoke
mock_agent = SimpleNamespace() mock_agent = SimpleNamespace()
mock_agent.ainvoke = AsyncMock(
return_value={"messages": [SimpleNamespace(content="Respuesta mock")]}
)
def mock_invoke(state, config=None):
return {"messages": [SimpleNamespace(content="Respuesta mock")]}
mock_agent.invoke = mock_invoke
# Sobrescribir la dependencia del agente en la app
app.dependency_overrides[get_agent] = lambda: mock_agent app.dependency_overrides[get_agent] = lambda: mock_agent
async with AsyncClient( async with AsyncClient(
transport=ASGITransport(app=app), transport=ASGITransport(app=app), base_url="http://localhost:8010"
base_url="http://localhost:8010"
) as client: ) as client:
response = await client.post("/webhook", json=send_message_payload) response = await client.post("/webhook", json=send_message_payload)
# Limpiar override para no afectar a otros tests
app.dependency_overrides.pop(get_agent, None) app.dependency_overrides.pop(get_agent, None)
assert response.status_code == 200 assert response.status_code == 200