from __future__ import annotations from typing import Any, Callable, Dict from fastapi import FastAPI, HTTPException from .models import ToolCallRequest, ToolCallResponse, ToolDefinition, ToolListResponse from .settings import get_settings from .store import ( consultar_cliente_backoffice, consultar_reclamacao, registrar_acao_backoffice, consultar_siebel_caso, registrar_acao_siebel, consultar_imdb_cliente, consultar_speech_analytics, consultar_tais_kb, consultar_abrt, consultar_portabilidade, buscar_templates_emulador, gerar_rascunho_emulador, ) app = FastAPI(title="Backoffice MCP HTTP Server", version="1.0.0") TOOLS: Dict[str, Dict[str, Any]] = { "consultar_reclamacao": { "description": "Consulta reclamação/protocolo de backoffice/ANATEL.", "handler": consultar_reclamacao, "input_schema": { "type": "object", "properties": { "protocol_id": {"type": "string"}, "customer_key": {"type": "string"}, "interaction_key": {"type": "string"}, }, "required": ["protocol_id"], }, }, "consultar_cliente_backoffice": { "description": "Consulta contexto operacional do cliente para backoffice.", "handler": consultar_cliente_backoffice, "input_schema": { "type": "object", "properties": { "customer_key": {"type": "string"}, "contract_key": {"type": "string"}, "session_key": {"type": "string"}, "cpf": {"type": "string"}, "cnpj": {"type": "string"}, "document": {"type": "string"}, "document_type": {"type": "string"}, }, "required": ["customer_key"], }, }, "registrar_acao_backoffice": { "description": "Registra ação operacional, parecer ou encaminhamento no sistema de backoffice.", "handler": registrar_acao_backoffice, "input_schema": { "type": "object", "properties": { "protocol_id": {"type": "string"}, "action_text": {"type": "string"}, "operator_session": {"type": "string"}, }, "required": ["protocol_id", "action_text"], }, }, "consultar_siebel_caso": { "description": "Consulta caso/SR no Siebel para reclamação ANATEL/backoffice.", "handler": consultar_siebel_caso, "input_schema": {"type": "object", "properties": {"protocol_id": {"type": "string"}, "interaction_key": {"type": "string"}, "customer_key": {"type": "string"}}}, }, "registrar_acao_siebel": { "description": "Registra ação, reclassificação, fechamento ou atualização no Siebel.", "handler": registrar_acao_siebel, "input_schema": {"type": "object", "properties": {"protocol_id": {"type": "string"}, "action_text": {"type": "string"}, "operator_session": {"type": "string"}}, "required": ["action_text"]}, }, "consultar_imdb_cliente": { "description": "Consulta enriquecimento IMDB/PMID do cliente.", "handler": consultar_imdb_cliente, "input_schema": {"type": "object", "properties": {"customer_key": {"type": "string"}, "contract_key": {"type": "string"}, "session_key": {"type": "string"}, "cpf": {"type": "string"}, "cnpj": {"type": "string"}, "document": {"type": "string"}, "document_type": {"type": "string"}}}, }, "consultar_speech_analytics": { "description": "Consulta histórico/resumo do Speech Analytics.", "handler": consultar_speech_analytics, "input_schema": {"type": "object", "properties": {"protocol_id": {"type": "string"}, "customer_key": {"type": "string"}, "interaction_key": {"type": "string"}, "cpf": {"type": "string"}, "cnpj": {"type": "string"}, "document": {"type": "string"}, "document_type": {"type": "string"}}}, }, "consultar_tais_kb": { "description": "Consulta base TAIS KB/RAG e templates de resposta.", "handler": consultar_tais_kb, "input_schema": {"type": "object", "properties": {"query": {"type": "string"}, "protocol_id": {"type": "string"}, "customer_key": {"type": "string"}}}, }, "consultar_abrt": { "description": "Consulta suporte ABRT associado ao cliente/caso.", "handler": consultar_abrt, "input_schema": {"type": "object", "properties": {"customer_key": {"type": "string"}, "protocol_id": {"type": "string"}, "cpf": {"type": "string"}, "cnpj": {"type": "string"}, "document": {"type": "string"}, "document_type": {"type": "string"}}}, }, "consultar_portabilidade": { "description": "Consulta status de portabilidade.", "handler": consultar_portabilidade, "input_schema": {"type": "object", "properties": {"customer_key": {"type": "string"}, "contract_key": {"type": "string"}, "cpf": {"type": "string"}, "cnpj": {"type": "string"}, "document": {"type": "string"}, "document_type": {"type": "string"}}}, }, "buscar_templates_emulador": { "description": "Busca templates/documentos para Response Emulator.", "handler": buscar_templates_emulador, "input_schema": {"type": "object", "properties": {"protocol_id": {"type": "string"}, "query": {"type": "string"}}}, }, "gerar_rascunho_emulador": { "description": "Gera rascunho de resposta do Response Emulator.", "handler": gerar_rascunho_emulador, "input_schema": {"type": "object", "properties": {"protocol_id": {"type": "string"}, "selected_actions": {"type": "array"}, "operator_instructions": {"type": "string"}}}, }, } @app.get("/health") def health() -> Dict[str, Any]: settings = get_settings() return { "status": "ok", "service": "backoffice_mcp_server", "use_mock": settings.use_mock, "backend_type": settings.backend_type, "rest_configured": bool(settings.rest_base_url), "oracle_configured": bool(settings.oracle_dsn and settings.oracle_user), "tim_clients_configured": settings.tim_clients_configured(), "tim_env": { "PMID_API_HOST": bool(settings.pmid_api_host), "SIEBEL_API_HOST": bool(settings.siebel_api_host), "SPEECH_PREDICTION_BASE_URL": bool(settings.speech_prediction_base_url), "SPEECH_HISTORY_BASE_URL": bool(settings.speech_history_base_url), "TAIS_DB_DSN": bool(settings.tais_db_dsn), "TAIS_GENAI_ENDPOINT": bool(settings.tais_genai_endpoint), }, "tools": list(TOOLS.keys()), } @app.get("/tools/list", response_model=ToolListResponse) def list_tools() -> ToolListResponse: return ToolListResponse( tools=[ ToolDefinition( name=name, description=cfg["description"], input_schema=cfg["input_schema"], ) for name, cfg in TOOLS.items() ] ) @app.post("/tools/call", response_model=ToolCallResponse) def call_tool(req: ToolCallRequest) -> ToolCallResponse: cfg = TOOLS.get(req.name) if not cfg: raise HTTPException(status_code=404, detail=f"Ferramenta não encontrada: {req.name}") handler: Callable[..., Dict[str, Any]] = cfg["handler"] try: schema_props = ((cfg.get("input_schema") or {}).get("properties") or {}) allowed = set(schema_props.keys()) # O MCPToolRouter do framework preserva original_context/extra_args por # design. O servidor de domínio deve aceitar somente os parâmetros da # tool, senão handlers Python falham com unexpected keyword argument. arguments = dict(req.arguments or {}) filtered_arguments = {k: v for k, v in arguments.items() if not allowed or k in allowed} result = handler(**filtered_arguments) return ToolCallResponse(ok=True, tool=req.name, result=result) except TypeError as exc: return ToolCallResponse(ok=False, tool=req.name, error=f"Parâmetros inválidos: {exc}") except Exception as exc: # pragma: no cover return ToolCallResponse(ok=False, tool=req.name, error=str(exc))