This commit is contained in:
a2a-platform
2026-06-04 22:35:23 +00:00
parent c2768d3117
commit d29e0ca80a
3 changed files with 2998 additions and 2 deletions

2526
.a2a/agent.dsl.json Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -8,3 +8,16 @@ Generated A2APack agent for Mailu API.
The generated code is intentionally editable. Tune prompts, operation The generated code is intentionally editable. Tune prompts, operation
grouping, auth names, and safety policy before publishing serious agents. grouping, auth names, and safety policy before publishing serious agents.
## Mailbox skills
This agent also includes SMTP/IMAP skills:
- `send_email`
- `read_inbox`
- `mark_email_read`
These skills take the mailbox email address, mint a temporary Mailu user token
through the admin API, use it as the SMTP/IMAP password, and delete the token
before returning. The token is not exposed to the model or returned in skill
responses.

461
agent.py
View File

@@ -1,8 +1,18 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import base64 import base64
from contextlib import asynccontextmanager
from email.header import decode_header, make_header
from email.message import EmailMessage
from email.parser import BytesParser
from email.policy import default as default_email_policy
from email.utils import formatdate, make_msgid, parsedate_to_datetime
import imaplib
import json import json
import re import re
import smtplib
import ssl
from typing import Any from typing import Any
import httpx import httpx
@@ -42,13 +52,44 @@ class OperationInput(BaseModel):
body: Any | None = Field(default=None, description="JSON request body, when the operation accepts one.") body: Any | None = Field(default=None, description="JSON request body, when the operation accepts one.")
class SendEmailInput(BaseModel):
email: str = Field(description="Mailbox email address to send as. A temporary Mailu token is minted internally.")
to: list[str] = Field(description="Recipient email addresses.")
subject: str = Field(description="Message subject.")
text_body: str = Field(description="Plain text message body.")
cc: list[str] = Field(default_factory=list, description="CC recipients.")
bcc: list[str] = Field(default_factory=list, description="BCC recipients.")
reply_to: str | None = Field(default=None, description="Optional Reply-To address.")
smtp_host: str = Field(default="mail.a2acloud.io", description="SMTP submission host.")
smtp_port: int = Field(default=587, description="SMTP port. Use 587 for STARTTLS or 465 for implicit TLS.")
class ReadInboxInput(BaseModel):
email: str = Field(description="Mailbox email address to read. A temporary Mailu token is minted internally.")
mailbox: str = Field(default="INBOX", description="IMAP mailbox name.")
limit: int = Field(default=10, ge=1, le=50, description="Maximum number of newest messages to return.")
unread_only: bool = Field(default=True, description="Only return unread messages.")
mark_seen: bool = Field(default=False, description="Mark returned messages as read.")
max_body_chars: int = Field(default=4000, ge=0, le=20000, description="Maximum plain text body characters per message.")
imap_host: str = Field(default="mail.a2acloud.io", description="IMAPS host.")
imap_port: int = Field(default=993, description="IMAPS port.")
class MarkEmailReadInput(BaseModel):
email: str = Field(description="Mailbox email address. A temporary Mailu token is minted internally.")
uid: str = Field(description="IMAP UID returned by read_inbox.")
mailbox: str = Field(default="INBOX", description="IMAP mailbox name.")
imap_host: str = Field(default="mail.a2acloud.io", description="IMAPS host.")
imap_port: int = Field(default=993, description="IMAPS port.")
class MailuOpenapiAgent(A2AAgent): class MailuOpenapiAgent(A2AAgent):
name = "mailu-openapi-agent" name = "mailu-openapi-agent"
description = "Mailu administration agent generated from the live Mailu OpenAPI spec." description = "Mailu administration agent generated from the live Mailu OpenAPI spec."
version = "1.0" version = "1.0"
consumer_setup = ConsumerSetup.from_fields( consumer_setup = ConsumerSetup.from_fields(
ConsumerSetupField.config("OPENAPI_BASE_URL", label="API base URL", description="Override the default API server (https://mail.a2acloud.io/api/v1).", required=False, input_type="url"), ConsumerSetupField.config("OPENAPI_BASE_URL", label="API base URL", description="Override the default API server (https://mail.a2acloud.io/api/v1).", required=False, input_type="url"),
ConsumerSetupField.secret("AUTHORIZATION", label="Bearer API key", description="Sent as header parameter 'Authorization'.", required=True), ConsumerSetupField.secret("AUTHORIZATION", label="Bearer API key", description="Sent as header parameter 'Authorization'. Also used to mint temporary per-user Mailu mailbox tokens for SMTP/IMAP skills.", required=True),
) )
llm_provisioning = LLMProvisioning.PLATFORM llm_provisioning = LLMProvisioning.PLATFORM
pricing = Pricing( pricing = Pricing(
@@ -114,6 +155,100 @@ class MailuOpenapiAgent(A2AAgent):
"messages": [_message_to_dict(message) for message in messages[-8:]], "messages": [_message_to_dict(message) for message in messages[-8:]],
} }
@skill(
name="send_email",
description="Mint a temporary Mailu token for a mailbox, send an email over SMTP submission, then delete the token.",
tags=("mailbox", "smtp", "email"),
timeout_seconds=120,
)
async def send_email(
self,
ctx: RunContext,
email: str,
to: list[str],
subject: str,
text_body: str,
cc: list[str] | None = None,
bcc: list[str] | None = None,
reply_to: str | None = None,
smtp_host: str = "mail.a2acloud.io",
smtp_port: int = 587,
) -> dict[str, Any]:
if not to:
return {"ok": False, "error": "recipient_required"}
async with self._temporary_mail_token(ctx, email, "send_email") as token_info:
return await asyncio.to_thread(
_send_email_sync,
email,
token_info["token"],
to,
subject,
text_body,
cc or [],
bcc or [],
reply_to,
smtp_host,
smtp_port,
)
@skill(
name="read_inbox",
description="Mint a temporary Mailu token for a mailbox, read messages over IMAPS, then delete the token.",
tags=("mailbox", "imap", "email"),
timeout_seconds=120,
)
async def read_inbox(
self,
ctx: RunContext,
email: str,
mailbox: str = "INBOX",
limit: int = 10,
unread_only: bool = True,
mark_seen: bool = False,
max_body_chars: int = 4000,
imap_host: str = "mail.a2acloud.io",
imap_port: int = 993,
) -> dict[str, Any]:
async with self._temporary_mail_token(ctx, email, "read_inbox") as token_info:
return await asyncio.to_thread(
_read_inbox_sync,
email,
token_info["token"],
mailbox,
limit,
unread_only,
mark_seen,
max_body_chars,
imap_host,
imap_port,
)
@skill(
name="mark_email_read",
description="Mint a temporary Mailu token for a mailbox, mark one IMAP UID as read, then delete the token.",
tags=("mailbox", "imap", "email"),
timeout_seconds=60,
)
async def mark_email_read(
self,
ctx: RunContext,
email: str,
uid: str,
mailbox: str = "INBOX",
imap_host: str = "mail.a2acloud.io",
imap_port: int = 993,
) -> dict[str, Any]:
async with self._temporary_mail_token(ctx, email, "mark_email_read") as token_info:
return await asyncio.to_thread(
_mark_email_read_sync,
email,
token_info["token"],
uid,
mailbox,
imap_host,
imap_port,
)
@skill( @skill(
name="create_alias", name="create_alias",
description="POST /alias - Create a new alias", description="POST /alias - Create a new alias",
@@ -837,7 +972,88 @@ class MailuOpenapiAgent(A2AAgent):
) )
def _operation_tools(self, ctx: RunContext) -> list[StructuredTool]: def _operation_tools(self, ctx: RunContext) -> list[StructuredTool]:
tools: list[StructuredTool] = [] async def send_email_tool(
email: str,
to: list[str],
subject: str,
text_body: str,
cc: list[str] | None = None,
bcc: list[str] | None = None,
reply_to: str | None = None,
smtp_host: str = "mail.a2acloud.io",
smtp_port: int = 587,
) -> dict[str, Any]:
return await self.send_email(
ctx,
email=email,
to=to,
subject=subject,
text_body=text_body,
cc=cc,
bcc=bcc,
reply_to=reply_to,
smtp_host=smtp_host,
smtp_port=smtp_port,
)
async def read_inbox_tool(
email: str,
mailbox: str = "INBOX",
limit: int = 10,
unread_only: bool = True,
mark_seen: bool = False,
max_body_chars: int = 4000,
imap_host: str = "mail.a2acloud.io",
imap_port: int = 993,
) -> dict[str, Any]:
return await self.read_inbox(
ctx,
email=email,
mailbox=mailbox,
limit=limit,
unread_only=unread_only,
mark_seen=mark_seen,
max_body_chars=max_body_chars,
imap_host=imap_host,
imap_port=imap_port,
)
async def mark_email_read_tool(
email: str,
uid: str,
mailbox: str = "INBOX",
imap_host: str = "mail.a2acloud.io",
imap_port: int = 993,
) -> dict[str, Any]:
return await self.mark_email_read(
ctx,
email=email,
uid=uid,
mailbox=mailbox,
imap_host=imap_host,
imap_port=imap_port,
)
tools: list[StructuredTool] = [
StructuredTool.from_function(
coroutine=send_email_tool,
name="send_email",
description="Send email through Mailu SMTP. Provide the mailbox email; the tool mints and deletes a temporary Mailu user token internally.",
args_schema=SendEmailInput,
),
StructuredTool.from_function(
coroutine=read_inbox_tool,
name="read_inbox",
description="Read Mailu mailbox messages through IMAPS. Provide the mailbox email; the tool mints and deletes a temporary Mailu user token internally.",
args_schema=ReadInboxInput,
),
StructuredTool.from_function(
coroutine=mark_email_read_tool,
name="mark_email_read",
description="Mark a Mailu mailbox message as read by IMAP UID. Provide the mailbox email; the tool mints and deletes a temporary Mailu user token internally.",
args_schema=MarkEmailReadInput,
),
]
for operation_id, operation in OPERATIONS.items(): for operation_id, operation in OPERATIONS.items():
async def call( async def call(
parameters: dict[str, Any] | None = None, parameters: dict[str, Any] | None = None,
@@ -867,9 +1083,15 @@ class MailuOpenapiAgent(A2AAgent):
"You operate an API through generated OpenAPI tools.", "You operate an API through generated OpenAPI tools.",
"Call tools to get real results. Do not invent API responses.", "Call tools to get real results. Do not invent API responses.",
"For write, update, or delete operations, explain the intended action before calling the tool.", "For write, update, or delete operations, explain the intended action before calling the tool.",
"For mailbox tasks, use send_email, read_inbox, and mark_email_read. These tools mint temporary Mailu user tokens internally; never ask the user for or reveal mailbox tokens.",
"If an operation reports missing consumer setup, tell the user which setup field is required.", "If an operation reports missing consumer setup, tell the user which setup field is required.",
"If an operation returns 404, 405, 410, or a schema/validation error that suggests the live API no longer matches these tools, tell the user this generated agent may need to be refreshed from the latest OpenAPI spec and ask whether they want to refresh it.", "If an operation returns 404, 405, 410, or a schema/validation error that suggests the live API no longer matches these tools, tell the user this generated agent may need to be refreshed from the latest OpenAPI spec and ask whether they want to refresh it.",
"", "",
"Mailbox tools:",
"- send_email: WRITE SMTP submission using a temporary per-user token.",
"- read_inbox: READ IMAPS mailbox messages using a temporary per-user token.",
"- mark_email_read: WRITE IMAP seen flag using a temporary per-user token.",
"",
"Available operations:", "Available operations:",
] ]
for operation in OPERATIONS.values(): for operation in OPERATIONS.values():
@@ -896,6 +1118,34 @@ class MailuOpenapiAgent(A2AAgent):
parts.append("Accepts a JSON request body.") parts.append("Accepts a JSON request body.")
return "\n".join(part for part in parts if part) return "\n".join(part for part in parts if part)
@asynccontextmanager
async def _temporary_mail_token(
self,
ctx: RunContext,
email_address: str,
purpose: str,
):
created = await self._request(
ctx,
"create_token_2",
parameters={"email": email_address},
body={"comment": f"mailu-openapi-agent temporary token for {purpose}"},
)
if not created.get("ok"):
raise RuntimeError(f"failed to mint temporary Mailu token: {created}")
token_info = _extract_mailu_token(created.get("result"))
try:
yield token_info
finally:
token_id = token_info.get("id")
if token_id:
await self._request(
ctx,
"delete_token",
parameters={"token_id": token_id},
body=None,
)
async def _request( async def _request(
self, self,
ctx: RunContext, ctx: RunContext,
@@ -1050,6 +1300,213 @@ def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str:
return "" return ""
def _extract_mailu_token(payload: Any) -> dict[str, str]:
if not isinstance(payload, dict):
raise RuntimeError("Mailu token response was not a JSON object")
token = payload.get("token")
token_id = payload.get("id")
if not token:
nested = payload.get("result")
if isinstance(nested, dict):
token = nested.get("token")
token_id = token_id or nested.get("id")
if not token:
raise RuntimeError("Mailu token response did not include a token")
return {"token": str(token), "id": str(token_id) if token_id is not None else ""}
def _send_email_sync(
email_address: str,
token: str,
to: list[str],
subject: str,
text_body: str,
cc: list[str],
bcc: list[str],
reply_to: str | None,
smtp_host: str,
smtp_port: int,
) -> dict[str, Any]:
recipients = [addr for addr in [*to, *cc, *bcc] if addr]
if not recipients:
return {"ok": False, "error": "recipient_required"}
message = EmailMessage()
message["From"] = email_address
message["To"] = ", ".join(to)
if cc:
message["Cc"] = ", ".join(cc)
if reply_to:
message["Reply-To"] = reply_to
message["Subject"] = subject
message["Date"] = formatdate(localtime=True)
message["Message-ID"] = make_msgid(domain=email_address.split("@")[-1])
message.set_content(text_body)
context = ssl.create_default_context()
if smtp_port == 465:
with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=30, context=context) as server:
server.login(email_address, token)
refused = server.send_message(message, from_addr=email_address, to_addrs=recipients)
else:
with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server:
server.ehlo()
server.starttls(context=context)
server.ehlo()
server.login(email_address, token)
refused = server.send_message(message, from_addr=email_address, to_addrs=recipients)
return {
"ok": not bool(refused),
"sent": not bool(refused),
"from": email_address,
"to": to,
"cc": cc,
"refused_recipients": refused,
}
def _read_inbox_sync(
email_address: str,
token: str,
mailbox: str,
limit: int,
unread_only: bool,
mark_seen: bool,
max_body_chars: int,
imap_host: str,
imap_port: int,
) -> dict[str, Any]:
limit = max(1, min(limit, 50))
messages: list[dict[str, Any]] = []
with imaplib.IMAP4_SSL(imap_host, imap_port, timeout=30) as imap:
imap.login(email_address, token)
status, _ = imap.select(mailbox, readonly=not mark_seen)
if status != "OK":
return {"ok": False, "error": "mailbox_select_failed", "mailbox": mailbox}
criteria = "UNSEEN" if unread_only else "ALL"
status, data = imap.uid("search", None, criteria)
if status != "OK":
return {"ok": False, "error": "search_failed", "mailbox": mailbox}
uids = (data[0] or b"").split()
for uid_bytes in reversed(uids[-limit:]):
uid = uid_bytes.decode("ascii", errors="replace")
fetch_item = "(FLAGS BODY[])" if mark_seen else "(FLAGS BODY.PEEK[])"
status, fetched = imap.uid("fetch", uid, fetch_item)
if status != "OK":
messages.append({"uid": uid, "ok": False, "error": "fetch_failed"})
continue
raw_message = _first_fetch_bytes(fetched)
if raw_message is None:
messages.append({"uid": uid, "ok": False, "error": "message_missing"})
continue
parsed = BytesParser(policy=default_email_policy).parsebytes(raw_message)
flags = _fetch_flags(fetched)
messages.append(_message_summary(uid, parsed, flags, max_body_chars))
return {
"ok": True,
"mailbox": mailbox,
"email": email_address,
"count": len(messages),
"messages": messages,
}
def _mark_email_read_sync(
email_address: str,
token: str,
uid: str,
mailbox: str,
imap_host: str,
imap_port: int,
) -> dict[str, Any]:
with imaplib.IMAP4_SSL(imap_host, imap_port, timeout=30) as imap:
imap.login(email_address, token)
status, _ = imap.select(mailbox)
if status != "OK":
return {"ok": False, "error": "mailbox_select_failed", "mailbox": mailbox, "uid": uid}
status, data = imap.uid("store", uid, "+FLAGS", "(\\Seen)")
return {
"ok": status == "OK",
"mailbox": mailbox,
"uid": uid,
"result": [part.decode("utf-8", errors="replace") if isinstance(part, bytes) else part for part in data],
}
def _first_fetch_bytes(fetched: list[Any]) -> bytes | None:
for item in fetched:
if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
return item[1]
return None
def _fetch_flags(fetched: list[Any]) -> list[str]:
out: list[str] = []
for item in fetched:
chunks = item if isinstance(item, tuple) else (item,)
for chunk in chunks:
if not isinstance(chunk, bytes):
continue
match = re.search(rb"FLAGS \(([^)]*)\)", chunk)
if match:
out.extend(flag.decode("ascii", errors="replace") for flag in match.group(1).split())
return sorted(set(out))
def _message_summary(uid: str, message: EmailMessage, flags: list[str], max_body_chars: int) -> dict[str, Any]:
date_value = message.get("date")
parsed_date = None
if date_value:
try:
parsed_date = parsedate_to_datetime(date_value).isoformat()
except (TypeError, ValueError):
parsed_date = str(date_value)
body = _plain_text_body(message)
if max_body_chars >= 0:
body = body[:max_body_chars]
return {
"uid": uid,
"message_id": message.get("message-id"),
"subject": _decode_header_value(message.get("subject")),
"from": _decode_header_value(message.get("from")),
"to": _decode_header_value(message.get("to")),
"date": parsed_date,
"flags": flags,
"seen": "\\Seen" in flags,
"body": body,
}
def _plain_text_body(message: EmailMessage) -> str:
if message.is_multipart():
for part in message.walk():
if part.get_content_type() == "text/plain" and not part.get_filename():
return _part_text(part)
for part in message.walk():
if part.get_content_type() == "text/html" and not part.get_filename():
return _part_text(part)
return ""
return _part_text(message)
def _part_text(part: EmailMessage) -> str:
try:
content = part.get_content()
except (LookupError, UnicodeDecodeError):
payload = part.get_payload(decode=True) or b""
charset = part.get_content_charset() or "utf-8"
return payload.decode(charset, errors="replace")
return content if isinstance(content, str) else str(content)
def _decode_header_value(value: str | None) -> str | None:
if value is None:
return None
try:
return str(make_header(decode_header(value)))
except (LookupError, UnicodeDecodeError, ValueError):
return value
def _message_text(message: Any) -> Any: def _message_text(message: Any) -> Any:
content = getattr(message, "content", message) content = getattr(message, "content", message)
return content return content