From 767f076048f0db3babe4dda394eae51ed867504c Mon Sep 17 00:00:00 2001 From: Yigit Colakoglu Date: Thu, 1 Jan 2026 15:24:06 -0800 Subject: [PATCH] Add tool call logging and reply fields --- README.md | 23 +++- src/server.py | 2 + src/services/email_service.py | 235 +++++++++++++++++++++++++--------- src/tools/calendar_tools.py | 7 + src/tools/contacts_tools.py | 7 + src/tools/email_tools.py | 118 +++++++++-------- src/tools/logging_utils.py | 60 +++++++++ 7 files changed, 335 insertions(+), 117 deletions(-) create mode 100644 src/tools/logging_utils.py diff --git a/README.md b/README.md index c2b6b02..c666d7b 100644 --- a/README.md +++ b/README.md @@ -145,11 +145,32 @@ Add your MCP endpoint at https://poke.com/settings/connections. | Category | Tools | | --- | --- | -| Email | `list_mailboxes`, `list_emails`, `list_drafts`, `read_email`, `search_emails`, `move_email`, `delete_email`, `delete_draft`, `save_draft`, `edit_draft`, `send_email`, `reply_email`, `set_email_flags`, `unsubscribe_email` | +| Email | `list_mailboxes`, `list_emails`, `list_drafts`, `read_email`, `search_emails`, `move_email`, `delete_email`, `delete_draft`, `save_draft`, `edit_draft`, `send_email`, `set_email_flags`, `unsubscribe_email` | | Calendar | `list_calendars`, `list_events`, `get_event`, `create_event`, `update_event`, `delete_event` | | Contacts | `list_addressbooks`, `list_contacts`, `get_contact`, `create_contact`, `update_contact`, `delete_contact` | | System | `get_server_info` | +### Replying to an email + +Use `reply_to_email_id` on `save_draft`, `edit_draft`, or `send_email` to create a reply without a separate tool. + +- Provide `reply_to_email_id` (and optionally `reply_mailbox`, default `INBOX`). +- `reply_all=true` includes original recipients; otherwise it replies to the sender/Reply-To. +- If `to`/`subject` are omitted, they are derived from the original email; `body` is still required. + +Example (send a reply): +```json +{ + "tool": "send_email", + "args": { + "reply_to_email_id": "12345", + "reply_mailbox": "INBOX", + "reply_all": true, + "body": "Thanks — sounds good to me." + } +} +``` + ## Database and Migrations The server uses SQLite (default: `/data/cache.db`) and Alembic. diff --git a/src/server.py b/src/server.py index be300d8..78efd10 100644 --- a/src/server.py +++ b/src/server.py @@ -19,6 +19,7 @@ from fastmcp import FastMCP from config import settings from database import init_db, close_db +from tools.logging_utils import log_tool_call # Configure logging logging.basicConfig( @@ -104,6 +105,7 @@ def register_tools(): # Server info tool (always available) @mcp.tool(description="Get information about this PIM MCP server including enabled services and version.") +@log_tool_call def get_server_info() -> dict: """Get server information and status.""" return { diff --git a/src/services/email_service.py b/src/services/email_service.py index e3fafe9..2af7e06 100644 --- a/src/services/email_service.py +++ b/src/services/email_service.py @@ -433,17 +433,49 @@ class EmailService: def save_draft( self, - to: list[str], - subject: str, - body: str, + to: Optional[list[str]] = None, + subject: Optional[str] = None, + body: Optional[str] = None, cc: Optional[list[str]] = None, bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, mailbox: Optional[str] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> OperationResult: try: draft_mailbox = mailbox or self._find_drafts_folder() or "Drafts" + if reply_to_email_id: + context, error = self._get_reply_context( + reply_mailbox or "INBOX", + reply_to_email_id, + reply_all, + cc, + self.settings.smtp_from_email, + ) + if error: + return error + if not to: + to = context["to"] + if subject is None: + subject = context["subject"] + if cc is None: + cc = context["cc"] + in_reply_to = context["in_reply_to"] + references = context["references"] + else: + in_reply_to = None + references = None + + if not to: + return OperationResult(success=False, message="'to' is required for drafts") + if subject is None: + return OperationResult(success=False, message="'subject' is required for drafts") + if body is None: + return OperationResult(success=False, message="'body' is required for drafts") + msg = self._build_draft_message( to=to, subject=subject, @@ -452,6 +484,8 @@ class EmailService: bcc=bcc, reply_to=reply_to, html_body=html_body, + in_reply_to=in_reply_to, + references=references, ) client = self._get_imap_client() append_result = client.append( @@ -481,6 +515,9 @@ class EmailService: bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> OperationResult: draft_mailbox = mailbox or self._find_drafts_folder() or "Drafts" try: @@ -502,6 +539,28 @@ class EmailService: resolved_body = body if body is not None else (existing.body_text or "") resolved_html = html_body if html_body is not None else existing.body_html + if reply_to_email_id: + context, error = self._get_reply_context( + reply_mailbox or "INBOX", + reply_to_email_id, + reply_all, + cc, + self.settings.smtp_from_email, + ) + if error: + return error + if to is None: + resolved_to = context["to"] + if subject is None: + resolved_subject = context["subject"] + if cc is None: + resolved_cc = context["cc"] + in_reply_to = context["in_reply_to"] + references = context["references"] + else: + in_reply_to = None + references = None + try: client = self._get_imap_client() client.select_folder(draft_mailbox) @@ -517,6 +576,8 @@ class EmailService: bcc=resolved_bcc, reply_to=reply_to, html_body=resolved_html, + in_reply_to=in_reply_to, + references=references, ) append_result = client.append( draft_mailbox, @@ -536,9 +597,9 @@ class EmailService: async def send_email( self, - to: list[str], - subject: str, - body: str, + to: Optional[list[str]] = None, + subject: Optional[str] = None, + body: Optional[str] = None, cc: Optional[list[str]] = None, bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, @@ -547,8 +608,38 @@ class EmailService: sender_name: Optional[str] = None, in_reply_to: Optional[str] = None, references: Optional[list[str]] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> OperationResult: try: + if reply_to_email_id: + resolved_name, resolved_email = self._resolve_sender(sender_email, sender_name) + context, error = self._get_reply_context( + reply_mailbox or "INBOX", + reply_to_email_id, + reply_all, + cc, + resolved_email, + ) + if error: + return error + if not to: + to = context["to"] + if subject is None: + subject = context["subject"] + if cc is None: + cc = context["cc"] + in_reply_to = context["in_reply_to"] + references = context["references"] + + if not to: + return OperationResult(success=False, message="'to' is required to send email") + if subject is None: + return OperationResult(success=False, message="'subject' is required to send email") + if body is None: + return OperationResult(success=False, message="'body' is required to send email") + msg = MIMEMultipart("alternative") msg["Subject"] = subject resolved_name, resolved_email = self._resolve_sender(sender_email, sender_name) @@ -609,67 +700,17 @@ class EmailService: sender_email: Optional[str] = None, sender_name: Optional[str] = None, ) -> OperationResult: - original = self.read_email(mailbox, email_id, format="both") - if not original: - return OperationResult( - success=False, - message=f"Email {email_id} not found in {mailbox}", - id=email_id, - ) - - resolved_name, resolved_email = self._resolve_sender(sender_email, sender_name) - reply_to_header = original.headers.get("Reply-To") - reply_to_email = None - if reply_to_header: - _, reply_to_email = parseaddr(reply_to_header) - if not reply_to_email: - reply_to_email = original.from_address.email - - to = [reply_to_email] if reply_to_email else [] - reply_cc: list[str] = [] - - if reply_all: - for addr in original.to_addresses + original.cc_addresses: - if addr.email and addr.email not in to: - reply_cc.append(addr.email) - - if cc: - reply_cc.extend(cc) - - to = self._dedupe_emails(to, resolved_email) - reply_cc = self._dedupe_emails(reply_cc, resolved_email) - - if not to and reply_cc: - to = [reply_cc.pop(0)] - - if not to: - return OperationResult( - success=False, - message="No valid recipients found for reply", - id=email_id, - ) - - subject = original.subject or "(No Subject)" - if not subject.lower().startswith("re:"): - subject = f"Re: {subject}" - - in_reply_to = original.headers.get("Message-ID") or original.in_reply_to - references = list(original.references) - if in_reply_to and in_reply_to not in references: - references.append(in_reply_to) - return await self.send_email( - to=to, - subject=subject, body=body, - cc=reply_cc or None, bcc=bcc, reply_to=reply_to, html_body=html_body, - sender_email=resolved_email, - sender_name=resolved_name, - in_reply_to=in_reply_to, - references=references or None, + sender_email=sender_email, + sender_name=sender_name, + reply_to_email_id=email_id, + reply_mailbox=mailbox, + reply_all=reply_all, + cc=cc, ) def _parse_envelope_addresses(self, addresses) -> list[EmailAddress]: @@ -805,6 +846,8 @@ class EmailService: bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, + in_reply_to: Optional[str] = None, + references: Optional[list[str]] = None, ) -> MIMEMultipart: msg = MIMEMultipart("alternative") msg["Subject"] = subject @@ -821,12 +864,80 @@ class EmailService: msg["Bcc"] = ", ".join(bcc) if reply_to: msg["Reply-To"] = reply_to + if in_reply_to: + msg["In-Reply-To"] = in_reply_to + if references: + msg["References"] = " ".join(references) msg.attach(MIMEText(body or "", "plain", "utf-8")) if html_body: msg.attach(MIMEText(html_body, "html", "utf-8")) return msg + def _get_reply_context( + self, + mailbox: str, + email_id: str, + reply_all: bool, + cc: Optional[list[str]], + sender_email: Optional[str], + ) -> tuple[dict, Optional[OperationResult]]: + original = self.read_email(mailbox, email_id, format="both") + if not original: + return {}, OperationResult( + success=False, + message=f"Email {email_id} not found in {mailbox}", + id=email_id, + ) + + reply_to_header = original.headers.get("Reply-To") + reply_to_email = None + if reply_to_header: + _, reply_to_email = parseaddr(reply_to_header) + if not reply_to_email: + reply_to_email = original.from_address.email + + to = [reply_to_email] if reply_to_email else [] + reply_cc: list[str] = [] + + if reply_all: + for addr in original.to_addresses + original.cc_addresses: + if addr.email and addr.email not in to: + reply_cc.append(addr.email) + + if cc: + reply_cc.extend(cc) + + to = self._dedupe_emails(to, sender_email) + reply_cc = self._dedupe_emails(reply_cc, sender_email) + + if not to and reply_cc: + to = [reply_cc.pop(0)] + + if not to: + return {}, OperationResult( + success=False, + message="No valid recipients found for reply", + id=email_id, + ) + + subject = original.subject or "(No Subject)" + if not subject.lower().startswith("re:"): + subject = f"Re: {subject}" + + in_reply_to = original.headers.get("Message-ID") or original.in_reply_to + references = list(original.references) + if in_reply_to and in_reply_to not in references: + references.append(in_reply_to) + + return { + "to": to, + "cc": reply_cc or None, + "subject": subject, + "in_reply_to": in_reply_to, + "references": references or None, + }, None + def _resolve_sender( self, sender_email: Optional[str], sender_name: Optional[str] ) -> tuple[Optional[str], str]: diff --git a/src/tools/calendar_tools.py b/src/tools/calendar_tools.py index 20b8c52..75f263f 100644 --- a/src/tools/calendar_tools.py +++ b/src/tools/calendar_tools.py @@ -2,18 +2,21 @@ from typing import Optional from fastmcp import FastMCP from services.calendar_service import CalendarService +from tools.logging_utils import log_tool_call def register_calendar_tools(mcp: FastMCP, service: CalendarService): """Register all calendar-related MCP tools.""" @mcp.tool(description="List all available calendars from CalDAV and configured ICS feeds. Returns calendar ID, name, and properties.") + @log_tool_call def list_calendars() -> list[dict]: """List all calendars.""" calendars = service.list_calendars() return [c.model_dump() for c in calendars] @mcp.tool(description="List events within a date range. If no calendar is specified, lists events from all calendars.") + @log_tool_call def list_events( start_date: str, end_date: str, @@ -59,6 +62,7 @@ def register_calendar_tools(mcp: FastMCP, service: CalendarService): } @mcp.tool(description="Get detailed information about a specific calendar event including attendees and recurrence.") + @log_tool_call def get_event( calendar_id: str, event_id: str, @@ -74,6 +78,7 @@ def register_calendar_tools(mcp: FastMCP, service: CalendarService): return result.model_dump() if result else None @mcp.tool(description="Create a new calendar event with title, time, location, attendees, and optional recurrence.") + @log_tool_call def create_event( calendar_id: str, title: str, @@ -105,6 +110,7 @@ def register_calendar_tools(mcp: FastMCP, service: CalendarService): return result.model_dump() @mcp.tool(description="Update an existing calendar event. Only provided fields will be modified.") + @log_tool_call def update_event( calendar_id: str, event_id: str, @@ -134,6 +140,7 @@ def register_calendar_tools(mcp: FastMCP, service: CalendarService): return result.model_dump() if result else None @mcp.tool(description="Delete a calendar event by ID.") + @log_tool_call def delete_event( calendar_id: str, event_id: str, diff --git a/src/tools/contacts_tools.py b/src/tools/contacts_tools.py index d110848..931dba5 100644 --- a/src/tools/contacts_tools.py +++ b/src/tools/contacts_tools.py @@ -2,18 +2,21 @@ from typing import Optional from fastmcp import FastMCP from services.contacts_service import ContactsService +from tools.logging_utils import log_tool_call def register_contacts_tools(mcp: FastMCP, service: ContactsService): """Register all contacts-related MCP tools.""" @mcp.tool(description="List all available address books from the CardDAV server.") + @log_tool_call def list_addressbooks() -> list[dict]: """List all address books.""" addressbooks = service.list_addressbooks() return [a.model_dump() for a in addressbooks] @mcp.tool(description="List contacts in an address book with optional search filtering and pagination.") + @log_tool_call def list_contacts( addressbook_id: str, search: Optional[str] = None, @@ -33,6 +36,7 @@ def register_contacts_tools(mcp: FastMCP, service: ContactsService): return result.model_dump() @mcp.tool(description="Get detailed information about a specific contact including all fields.") + @log_tool_call def get_contact( addressbook_id: str, contact_id: str, @@ -48,6 +52,7 @@ def register_contacts_tools(mcp: FastMCP, service: ContactsService): return result.model_dump() if result else None @mcp.tool(description="Create a new contact with name, emails, phones, addresses, and other details.") + @log_tool_call def create_contact( addressbook_id: str, first_name: Optional[str] = None, @@ -93,6 +98,7 @@ def register_contacts_tools(mcp: FastMCP, service: ContactsService): return result.model_dump() @mcp.tool(description="Update an existing contact. Only provided fields will be modified.") + @log_tool_call def update_contact( addressbook_id: str, contact_id: str, @@ -138,6 +144,7 @@ def register_contacts_tools(mcp: FastMCP, service: ContactsService): return result.model_dump() if result else None @mcp.tool(description="Delete a contact from an address book.") + @log_tool_call def delete_contact( addressbook_id: str, contact_id: str, diff --git a/src/tools/email_tools.py b/src/tools/email_tools.py index b0de1cb..8edc3dc 100644 --- a/src/tools/email_tools.py +++ b/src/tools/email_tools.py @@ -2,18 +2,21 @@ from typing import Optional from fastmcp import FastMCP from services.email_service import EmailService +from tools.logging_utils import log_tool_call def register_email_tools(mcp: FastMCP, service: EmailService): """Register all email-related MCP tools.""" @mcp.tool(description="List all mailboxes/folders in the email account. Returns name, path, message count, and unread count for each mailbox.") + @log_tool_call def list_mailboxes() -> list[dict]: """List all IMAP mailboxes/folders.""" mailboxes = service.list_mailboxes() return [m.model_dump() for m in mailboxes] @mcp.tool(description="List emails in a mailbox with pagination. Returns email summaries including subject, from, date, and read status.") + @log_tool_call def list_emails( mailbox: str = "INBOX", limit: int = 50, @@ -33,6 +36,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="List draft emails in the Drafts mailbox with pagination.") + @log_tool_call def list_drafts( mailbox: Optional[str] = None, limit: int = 50, @@ -52,6 +56,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Read a specific email by ID with full body content and attachment information.") + @log_tool_call def read_email( mailbox: str, email_id: str, @@ -69,6 +74,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() if result else None @mcp.tool(description="Search emails in a mailbox using various criteria like subject, sender, or body content.") + @log_tool_call def search_emails( query: str, mailbox: str = "INBOX", @@ -94,6 +100,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Move an email from one mailbox/folder to another.") + @log_tool_call def move_email( email_id: str, source_mailbox: str, @@ -111,6 +118,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Delete an email, either moving it to trash or permanently deleting it.") + @log_tool_call def delete_email( email_id: str, mailbox: str, @@ -128,6 +136,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Delete a drafted email by ID, optionally permanently.") + @log_tool_call def delete_draft( email_id: str, mailbox: Optional[str] = None, @@ -145,33 +154,53 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Save a new draft email to the Drafts mailbox.") + @log_tool_call def save_draft( - to: list[str], - subject: str, - body: str, + to: Optional[list[str]] = None, + subject: Optional[str] = None, + body: Optional[str] = None, cc: Optional[list[str]] = None, bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, mailbox: Optional[str] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> dict: """ Save a new email draft. Args: - to: List of recipient email addresses - subject: Email subject line - body: Plain text email body + to: List of recipient email addresses (required unless reply_to_email_id is set) + subject: Email subject line (required unless reply_to_email_id is set) + body: Plain text email body (required unless reply_to_email_id is set) cc: List of CC recipients (optional) bcc: List of BCC recipients (optional) reply_to: Reply-to address (optional) html_body: HTML version of the email body (optional) mailbox: Drafts mailbox/folder override (default: auto-detect) + reply_to_email_id: Email ID to reply to (optional) + reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX) + reply_all: Whether to include original recipients when replying (default: False) """ - result = service.save_draft(to, subject, body, cc, bcc, reply_to, html_body, mailbox) + result = service.save_draft( + to, + subject, + body, + cc, + bcc, + reply_to, + html_body, + mailbox, + reply_to_email_id, + reply_mailbox, + reply_all, + ) return result.model_dump() @mcp.tool(description="Edit an existing draft email. Only provided fields will be modified.") + @log_tool_call def edit_draft( email_id: str, mailbox: Optional[str] = None, @@ -182,6 +211,9 @@ def register_email_tools(mcp: FastMCP, service: EmailService): bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> dict: """ Update an existing draft email. @@ -196,6 +228,9 @@ def register_email_tools(mcp: FastMCP, service: EmailService): bcc: List of BCC recipients (optional) reply_to: Reply-to address (optional) html_body: HTML version of the email body (optional) + reply_to_email_id: Email ID to reply to (optional) + reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX) + reply_all: Whether to include original recipients when replying (default: False) """ result = service.update_draft( email_id, @@ -207,34 +242,44 @@ def register_email_tools(mcp: FastMCP, service: EmailService): bcc, reply_to, html_body, + reply_to_email_id, + reply_mailbox, + reply_all, ) return result.model_dump() @mcp.tool(description="Send a new email via SMTP. Supports plain text and HTML content, CC, BCC, reply-to, and custom sender.") + @log_tool_call async def send_email( - to: list[str], - subject: str, - body: str, + to: Optional[list[str]] = None, + subject: Optional[str] = None, + body: Optional[str] = None, cc: Optional[list[str]] = None, bcc: Optional[list[str]] = None, reply_to: Optional[str] = None, html_body: Optional[str] = None, sender_email: Optional[str] = None, sender_name: Optional[str] = None, + reply_to_email_id: Optional[str] = None, + reply_mailbox: Optional[str] = None, + reply_all: bool = False, ) -> dict: """ Send a new email. Args: - to: List of recipient email addresses - subject: Email subject line - body: Plain text email body + to: List of recipient email addresses (required unless reply_to_email_id is set) + subject: Email subject line (required unless reply_to_email_id is set) + body: Plain text email body (required unless reply_to_email_id is set) cc: List of CC recipients (optional) bcc: List of BCC recipients (optional) reply_to: Reply-to address (optional) html_body: HTML version of the email body (optional) sender_email: Sender email address (optional, defaults to SMTP_FROM_EMAIL) sender_name: Sender display name (optional, defaults to SMTP_FROM_NAME) + reply_to_email_id: Email ID to reply to (optional) + reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX) + reply_all: Whether to include original recipients when replying (default: False) """ result = await service.send_email( to, @@ -246,52 +291,16 @@ def register_email_tools(mcp: FastMCP, service: EmailService): html_body, sender_email, sender_name, - ) - return result.model_dump() - - @mcp.tool(description="Reply to an existing email by ID, with optional reply-all behavior.") - async def reply_email( - email_id: str, - body: str, - mailbox: str = "INBOX", - reply_all: bool = False, - cc: Optional[list[str]] = None, - bcc: Optional[list[str]] = None, - reply_to: Optional[str] = None, - html_body: Optional[str] = None, - sender_email: Optional[str] = None, - sender_name: Optional[str] = None, - ) -> dict: - """ - Reply to an existing email. - - Args: - email_id: The unique ID of the email to reply to - body: Plain text email body - mailbox: The mailbox containing the email (default: INBOX) - reply_all: Whether to include original recipients (default: False) - cc: List of CC recipients (optional) - bcc: List of BCC recipients (optional) - reply_to: Reply-to address for the reply (optional) - html_body: HTML version of the email body (optional) - sender_email: Sender email address (optional, defaults to SMTP_FROM_EMAIL) - sender_name: Sender display name (optional, defaults to SMTP_FROM_NAME) - """ - result = await service.reply_email( - mailbox, - email_id, - body, + None, + None, + reply_to_email_id, + reply_mailbox, reply_all, - cc, - bcc, - reply_to, - html_body, - sender_email, - sender_name, ) return result.model_dump() @mcp.tool(description="Set or remove IMAP flags on an email. Standard flags: \\Seen, \\Answered, \\Flagged, \\Deleted, \\Draft. Custom keywords are also supported.") + @log_tool_call def set_email_flags( email_id: str, mailbox: str, @@ -311,6 +320,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService): return result.model_dump() @mcp.tool(description="Unsubscribe from a mailing list. Parses List-Unsubscribe headers and attempts automatic unsubscribe via HTTP or provides mailto instructions.") + @log_tool_call async def unsubscribe_email( email_id: str, mailbox: str = "INBOX", diff --git a/src/tools/logging_utils.py b/src/tools/logging_utils.py new file mode 100644 index 0000000..a5b78bc --- /dev/null +++ b/src/tools/logging_utils.py @@ -0,0 +1,60 @@ +import functools +import inspect +import json +import logging +import os + +logger = logging.getLogger("mcp.tools") +_MAX_LOG_CHARS = int(os.getenv("TOOL_LOG_MAX_CHARS", "4000")) + + +def _serialize(value) -> str: + try: + return json.dumps(value, default=str, ensure_ascii=True) + except TypeError: + return repr(value) + + +def _truncate(text: str) -> str: + if len(text) <= _MAX_LOG_CHARS: + return text + truncated = len(text) - _MAX_LOG_CHARS + return f"{text[:_MAX_LOG_CHARS]}...(truncated {truncated} chars)" + + +def log_tool_call(func): + """Log tool calls and responses with optional truncation.""" + if inspect.iscoroutinefunction(func): + @functools.wraps(func) + async def wrapper(*args, **kwargs): + logger.info( + "Tool call %s args=%s kwargs=%s", + func.__name__, + _truncate(_serialize(args)), + _truncate(_serialize(kwargs)), + ) + result = await func(*args, **kwargs) + logger.info( + "Tool response %s result=%s", + func.__name__, + _truncate(_serialize(result)), + ) + return result + return wrapper + + @functools.wraps(func) + def wrapper(*args, **kwargs): + logger.info( + "Tool call %s args=%s kwargs=%s", + func.__name__, + _truncate(_serialize(args)), + _truncate(_serialize(kwargs)), + ) + result = func(*args, **kwargs) + logger.info( + "Tool response %s result=%s", + func.__name__, + _truncate(_serialize(result)), + ) + return result + return wrapper