fix(logging): configure stderr logging + tee capture, add debug confirmation
All checks were successful
Build and Push Docker Image / build (push) Successful in 37s

- Set up logging.basicConfig() at module load so scraper logs reach stderr
  (visible in docker logs instead of silently dropped)
- Replace StringIO-only capture with TeeHandler that writes to BOTH stderr
  and the StringIO buffer, so logs remain visible in docker while also
  being returned in tool responses
- Add explicit 'LOGIN TOOL CALLED' and 'DEBUG MODE ENABLED' log lines
  at the start of the login tool so users can verify logging is active
This commit is contained in:
2026-04-28 02:16:31 +00:00
parent 1999392df7
commit f51e61b8d7

View File

@@ -2,6 +2,7 @@ import io
import json import json
import logging import logging
import os import os
import sys
import time import time
from contextlib import contextmanager from contextlib import contextmanager
from typing import Optional, Any, Tuple from typing import Optional, Any, Tuple
@@ -16,33 +17,77 @@ import schwab_scraper.unified_api as api
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Log capture helper — scraper logs go to stderr which is invisible in MCP # Configure logging so it actually reaches stderr (visible in docker logs).
# stdio mode. This captures them so we can return them in the response. # The scraper and MCP libraries log extensively but don't set up handlers
# when imported as a module, so messages are silently dropped.
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
# Ensure the scraper logger propagates to our root handler
_scraper_logger = logging.getLogger("schwab_scraper")
_scraper_logger.setLevel(logging.DEBUG if os.getenv("SCHWAB_DEBUG", "").lower() in ("1", "true") else logging.INFO)
_scraper_logger.propagate = True
# ---------------------------------------------------------------------------
# Log capture helper — captures scraper logs to a string buffer AND tees
# them to stderr so they remain visible in docker logs.
# ---------------------------------------------------------------------------
class _TeeHandler(logging.StreamHandler):
"""Handler that copies every record to a secondary (StringIO) buffer."""
def __init__(self, stream, extra_buf: io.StringIO, level=logging.NOTSET):
super().__init__(stream)
self.extra_buf = extra_buf
self.tee_level = level
def emit(self, record):
super().emit(record)
if record.levelno >= self.tee_level:
try:
msg = self.format(record)
self.extra_buf.write(msg + "\n")
self.extra_buf.flush()
except Exception:
pass
@contextmanager @contextmanager
def capture_logs(logger_name: str = "schwab_scraper", level: int = logging.DEBUG): def capture_logs(logger_name: str = "schwab_scraper", level: int = logging.DEBUG):
"""Context manager that captures log output to a string buffer. """
Context manager that captures log output to a string buffer
while still writing to stderr (docker-visible).
Yields the buffer so callers can read captured logs after the block. Yields the buffer so callers can read captured logs after the block.
""" """
logger = logging.getLogger(logger_name) logger = logging.getLogger(logger_name)
# Ensure the logger will actually process messages at this level
old_level = logger.level old_level = logger.level
if old_level > level: if old_level > level:
logger.setLevel(level) logger.setLevel(level)
buf = io.StringIO() buf = io.StringIO()
handler = logging.StreamHandler(buf) handler = _TeeHandler(sys.stderr, buf, level=level)
handler.setLevel(level) handler.setLevel(level)
# Match the format used by the scraper
handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")) handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
logger.addHandler(handler) logger.addHandler(handler)
# Also tee the root logger in case scraper logs through sub-loggers
root_old_level = logging.getLogger().level
if root_old_level > level:
logging.getLogger().setLevel(level)
try: try:
yield buf yield buf
finally: finally:
logger.removeHandler(handler) logger.removeHandler(handler)
logger.setLevel(old_level) if old_level != logger.level:
logger.setLevel(old_level)
if root_old_level != logging.getLogger().level:
logging.getLogger().setLevel(root_old_level)
def _enrich_with_logs(result: dict, log_buffer: io.StringIO, debug: bool) -> dict: def _enrich_with_logs(result: dict, log_buffer: io.StringIO, debug: bool) -> dict:
@@ -218,11 +263,20 @@ async def login(
"data": None, "data": None,
}) })
mcp_logger = logging.getLogger("schwab_mcp_custom")
mcp_logger.info("=== LOGIN TOOL CALLED ===")
mcp_logger.info(f"debug={debug}, username_provided={bool(username)}")
with capture_logs(level=logging.DEBUG if debug else logging.INFO) as log_buf: with capture_logs(level=logging.DEBUG if debug else logging.INFO) as log_buf:
mcp_logger.info("capture_logs context entered")
if debug:
mcp_logger.info("DEBUG MODE ENABLED — verbose logging active")
result = await api.login(username=username, password=password, debug=debug) result = await api.login(username=username, password=password, debug=debug)
success = result.get("success", False) success = result.get("success", False)
login_manager.record_attempt(success) login_manager.record_attempt(success)
mcp_logger.info(f"api.login completed — success={success}")
result = _enrich_with_logs(result, log_buf, debug) result = _enrich_with_logs(result, log_buf, debug)
mcp_logger.info("capture_logs context exited, returning result")
return serialize(result) return serialize(result)