Read emails from IMAP mailbox. Args: config: Resolved email config limit: Max number of emails to return search: Optional IMAP search criteria (e.g. 'FROM "john"', 'SUBJECT "hello"') folder: Mailbox folder (default INBOX)
(
config: dict,
limit: int = 10,
search: Optional[str] = None,
folder: str = "INBOX",
)
| 255 | |
| 256 | |
| 257 | async def read_emails( |
| 258 | config: dict, |
| 259 | limit: int = 10, |
| 260 | search: Optional[str] = None, |
| 261 | folder: str = "INBOX", |
| 262 | ) -> str: |
| 263 | """Read emails from IMAP mailbox. |
| 264 | |
| 265 | Args: |
| 266 | config: Resolved email config |
| 267 | limit: Max number of emails to return |
| 268 | search: Optional IMAP search criteria (e.g. 'FROM "john"', 'SUBJECT "hello"') |
| 269 | folder: Mailbox folder (default INBOX) |
| 270 | """ |
| 271 | cfg = resolve_config(config) |
| 272 | addr = cfg["email_address"] |
| 273 | password = cfg["auth_code"] |
| 274 | |
| 275 | if not addr or not password: |
| 276 | return "❌ Email not configured. Please set email address and authorization code in tool config." |
| 277 | |
| 278 | limit = min(limit, 30) # Cap at 30 |
| 279 | |
| 280 | try: |
| 281 | with force_ipv4(): |
| 282 | context = ssl.create_default_context() |
| 283 | with imaplib.IMAP4_SSL(cfg["imap_host"], cfg["imap_port"], ssl_context=context) as mail: |
| 284 | mail.login(addr, password) |
| 285 | mail.select(folder, readonly=True) |
| 286 | |
| 287 | # Search |
| 288 | if search: |
| 289 | _, msg_nums = mail.search(None, search) |
| 290 | else: |
| 291 | _, msg_nums = mail.search(None, "ALL") |
| 292 | |
| 293 | msg_ids = msg_nums[0].split() |
| 294 | if not msg_ids: |
| 295 | return "📭 No emails found." |
| 296 | |
| 297 | # Get latest N emails |
| 298 | latest_ids = msg_ids[-limit:] |
| 299 | latest_ids.reverse() # Newest first |
| 300 | |
| 301 | results = [] |
| 302 | for mid in latest_ids: |
| 303 | _, msg_data = mail.fetch(mid, "(RFC822)") |
| 304 | if not msg_data or not msg_data[0]: |
| 305 | continue |
| 306 | raw = msg_data[0][1] |
| 307 | msg = email_lib.message_from_bytes(raw) |
| 308 | |
| 309 | from_addr = _decode_header_value(msg.get("From", "")) |
| 310 | subject = _decode_header_value(msg.get("Subject", "(No subject)")) |
| 311 | date_str = msg.get("Date", "") |
| 312 | message_id = msg.get("Message-ID", "") |
| 313 | body = _extract_body(msg) |
| 314 | # Truncate body for readability |
no test coverage detected