diff --git a/python/src/agentmail_toolkit/functions.py b/python/src/agentmail_toolkit/functions.py index 1fe4d80..336295a 100644 --- a/python/src/agentmail_toolkit/functions.py +++ b/python/src/agentmail_toolkit/functions.py @@ -4,6 +4,8 @@ from agentmail import AgentMail from agentmail.inboxes import CreateInboxRequest +from .schemas import CreateDomainParams + import io import logging import filetype @@ -52,6 +54,28 @@ def delete_inbox(client: AgentMail, kwargs: Kwargs): return client.inboxes.delete(**kwargs) +def list_domains(client: AgentMail, kwargs: Kwargs): + return client.domains.list(**kwargs) + + +def get_domain(client: AgentMail, kwargs: Kwargs): + return client.domains.get(**kwargs) + + +def create_domain(client: AgentMail, kwargs: Kwargs): + # Parsed so only the fields this tool offers reach the SDK, whose create also takes + # allow_conflicting_provider; the OpenAI and LiveKit adapters hand over raw arguments. + return client.domains.create(**CreateDomainParams(**kwargs).model_dump(exclude_none=True)) + + +def get_domain_setup_link(client: AgentMail, kwargs: Kwargs): + return client.domains.get_setup_link(**kwargs) + + +def verify_domain(client: AgentMail, kwargs: Kwargs): + return client.domains.verify(**kwargs) + + def list_threads(client: AgentMail, kwargs: Kwargs): return client.inboxes.threads.list(**kwargs) diff --git a/python/src/agentmail_toolkit/schemas.py b/python/src/agentmail_toolkit/schemas.py index eee7797..92ee837 100644 --- a/python/src/agentmail_toolkit/schemas.py +++ b/python/src/agentmail_toolkit/schemas.py @@ -6,6 +6,7 @@ ThreadIdField = Annotated[str, Field(description="ID of thread")] MessageIdField = Annotated[str, Field(description="ID of message")] AttachmentIdField = Annotated[str, Field(description="ID of attachment")] +DomainIdField = Annotated[str, Field(description="ID of domain (from list_domains or create_domain)")] class ListItemsParams(BaseModel): @@ -25,6 +26,28 @@ class CreateInboxParams(BaseModel): display_name: Optional[str] = Field(default=None, description="Display name") +class GetDomainParams(BaseModel): + domain_id: DomainIdField + + +# allow_conflicting_provider is not offered: every domain made here receives mail, so its MX record is +# required, and a domain cannot receive at AgentMail and at Google Workspace or Microsoft 365 at once. +# With the flag the agent could only stop the user's current mail or end up receiving nothing. +class CreateDomainParams(BaseModel): + domain: str = Field(description="Domain name, e.g. example.com or mail.example.com") + feedback_enabled: Optional[bool] = Field( + default=None, description="Send bounce and complaint notifications to your inboxes. Default true" + ) + subdomains_enabled: Optional[bool] = Field( + default=None, + description="Allow inboxes on any subdomain of this domain. Adds a required wildcard MX record. Default false", + ) + tracking_enabled: Optional[bool] = Field( + default=None, + description="Serve open tracking from this domain. Adds a required link CNAME record. Default false", + ) + + class ListInboxItemsParams(ListItemsParams): inbox_id: InboxIdField labels: Optional[List[str]] = Field(default=None, description="Filter items with labels") diff --git a/python/src/agentmail_toolkit/tools.py b/python/src/agentmail_toolkit/tools.py index a095a8e..a6e85f9 100644 --- a/python/src/agentmail_toolkit/tools.py +++ b/python/src/agentmail_toolkit/tools.py @@ -7,6 +7,8 @@ ListInboxItemsParams, GetInboxParams, CreateInboxParams, + GetDomainParams, + CreateDomainParams, GetThreadParams, GetAttachmentParams, SendMessageParams, @@ -22,6 +24,11 @@ get_inbox, create_inbox, delete_inbox, + list_domains, + get_domain, + create_domain, + get_domain_setup_link, + verify_domain, list_threads, get_thread, get_attachment, @@ -66,6 +73,69 @@ class Tool(BaseModel): params_schema=GetInboxParams, func=delete_inbox, ), + Tool( + name="list_domains", + description=( + "List custom email domains, paginated. Use get_domain for a domain's verification status and " + "DNS records. Requires the domain_read permission." + ), + params_schema=ListItemsParams, + func=list_domains, + ), + Tool( + name="get_domain", + description=( + "Get a custom email domain by ID: its verification status, the reason it is not verified yet, " + "and the DNS records to add at the domain's DNS provider, each with its own status and reason. " + "Requires the domain_read permission; inbox-scoped API keys cannot access domains." + ), + params_schema=GetDomainParams, + func=get_domain, + ), + Tool( + name="create_domain", + description=( + "Add a custom email domain. Returns it with status NOT_STARTED and the DNS records to add at the " + "domain's DNS provider. Next, call get_domain_setup_link for a one-click link that adds them, or " + "add them by hand, then call verify_domain. AgentMail receives the domain's mail, so adding its " + "MX record moves mail away from any provider the domain uses today. Fails with 422 when the " + "domain already receives mail through Google Workspace or Microsoft 365: suggest a subdomain such " + "as agents.example.com instead. Requires the domain_create permission." + ), + params_schema=CreateDomainParams, + func=create_domain, + ), + Tool( + name="get_domain_setup_link", + description=( + "Get a one-click link that adds a domain's DNS records at its DNS provider, when the provider " + "supports it (for example Cloudflare or Vercel). When supported is true, open url in a browser: " + "the domain owner signs in at provider_name, reviews the records and approves, and the provider " + "writes them; the browser then lands on the AgentMail console. When get_domain lists an MX " + "record, the link replaces the domain's current MX records, and mail to any provider the domain " + "uses today stops. conflicting_provider names such a provider when the check finds one, but the " + "check can miss it, so confirm with the user that the domain has no other mail provider before " + "opening such a link. When supported is false, add the records from get_domain by hand. Once the " + "records are in place, call verify_domain. Requires the domain_read permission." + ), + params_schema=GetDomainParams, + func=get_domain_setup_link, + ), + Tool( + name="verify_domain", + description=( + "Start verifying a domain once its DNS records are in place. Returns at once and the check runs " + "in the background: call get_domain to follow it until status is VERIFIED. Until then, the " + "domain's reason says what is left: a dns_records_* reason means a record is missing or wrong at " + "the DNS provider (each record has its own status and reason); ses_*_pending and " + "ses_*_temporary_failure clear on their own; ses_*_failed and ses_*_not_started need " + "verify_domain again once the records are right. DNS changes can take a while to be seen, so if " + "nothing changes after several checks, stop and tell the user the reason. Requires the " + "domain_update permission." + ), + params_schema=GetDomainParams, + func=verify_domain, + ), Tool( name="list_threads", description="List threads in inbox", diff --git a/python/tests/test_domains.py b/python/tests/test_domains.py new file mode 100644 index 0000000..cf0d89e --- /dev/null +++ b/python/tests/test_domains.py @@ -0,0 +1,450 @@ +"""The domain tools, driven through the real AgentMail SDK over a fake HTTP layer. + +A MagicMock client would accept any method name and any keyword, so it could not +catch a tool calling a method the SDK does not have, or sending a field the API +does not read. Here the SDK builds the real request and the test reads the method, +path, query and body it would have sent. +""" + +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest +from agentmail import AgentMail +from langchain_core.tools import ToolException as LangchainToolException +from livekit.agents import ToolError as LivekitToolError + +from agentmail_toolkit import functions +from agentmail_toolkit.schemas import CreateDomainParams, GetDomainParams +from agentmail_toolkit.tools import tools + +DOMAIN_TOOLS = ["list_domains", "get_domain", "create_domain", "get_domain_setup_link", "verify_domain"] + +SIGNED_URL = ( + "https://dash.cloudflare.com/domainconnect/v2/domainTemplates/providers/agentmail.to/services/email/apply" + "?domain=example.com&sig=SIG&key=dc1" +) + + +def domain_body(status="NOT_STARTED", record_status="MISSING"): + return { + "domain_id": "example.com", + "domain": "example.com", + "status": status, + "feedback_enabled": True, + "subdomains_enabled": False, + "tracking_enabled": False, + "records": [ + { + "type": "MX", + "name": "example.com", + "value": "inbound-smtp.us-east-1.amazonaws.com", + "status": record_status, + "priority": 10, + } + ], + "updated_at": "2026-07-10T12:00:00Z", + "created_at": "2026-07-10T12:00:00Z", + } + + +class FakeApi: + """Answers the domain routes the way the API does and records every request.""" + + def __init__(self, error=None): + self.requests = [] + self.error = error + self.verified = False + self.reads_after_verify = 0 + + def __call__(self, request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) if request.content else None + self.requests.append((request.method, request.url.path, dict(request.url.params), body)) + if self.error: + return httpx.Response(self.error[0], json=self.error[1]) + + path = request.url.path + if request.method == "GET" and path == "/v0/domains": + item = {k: v for k, v in domain_body().items() if k not in ("status", "records")} + return httpx.Response(200, json={"count": 1, "domains": [item]}) + if request.method == "POST" and path == "/v0/domains": + return httpx.Response(200, json=domain_body()) + if path.endswith("/setup-link"): + return httpx.Response( + 200, + json={"supported": True, "provider_name": "Cloudflare", "url": SIGNED_URL, "width": 750, "height": 750}, + ) + if path.endswith("/verify"): + self.verified = True + return httpx.Response(204) + if request.method == "GET" and path.startswith("/v0/domains/"): + if not self.verified: + return httpx.Response(200, json=domain_body()) + self.reads_after_verify += 1 + if self.reads_after_verify == 1: + return httpx.Response(200, json=domain_body("VERIFYING")) + return httpx.Response(200, json=domain_body("VERIFIED", "VALID")) + return httpx.Response(404, json={"name": "NotFoundError", "message": "Route not found"}) + + +def sdk(api: FakeApi) -> AgentMail: + return AgentMail(api_key="test-key", httpx_client=httpx.Client(transport=httpx.MockTransport(api))) + + +CONFLICT_422 = ( + 422, + { + "name": "UnprocessableEntityError", + "message": ( + 'Domain "example.com" is configured with Google Workspace. Domains with existing email providers ' + "cannot be used simultaneously with AgentMail. Please use a dedicated domain or subdomain." + ), + }, +) + + +# -------------------------------------------------------------------------- +# catalog +# -------------------------------------------------------------------------- + + +def by_name(name): + return next(t for t in tools if t.name == name) + + +def test_domain_tools_sit_after_delete_inbox_and_before_the_thread_tools(): + names = [t.name for t in tools] + start = names.index("delete_inbox") + 1 + assert names[start : start + len(DOMAIN_TOOLS)] == DOMAIN_TOOLS + assert names[start + len(DOMAIN_TOOLS)] == "list_threads" + + +@pytest.mark.parametrize( + "name, permission", + [ + ("list_domains", "domain_read"), + ("get_domain", "domain_read"), + ("create_domain", "domain_create"), + ("get_domain_setup_link", "domain_read"), + ("verify_domain", "domain_update"), + ], +) +def test_each_tool_names_the_permission_it_needs(name, permission): + assert f"Requires the {permission} permission" in by_name(name).description + + +def test_descriptions_use_the_python_field_names(): + for name in DOMAIN_TOOLS: + description = by_name(name).description + for camel in ("allowConflictingProvider", "conflictingProvider", "providerName", "domainId"): + assert camel not in description + + +def test_create_domain_explains_the_conflict_and_offers_no_override(): + description = by_name("create_domain").description + assert "422" in description + assert "Google Workspace or Microsoft 365" in description + assert "subdomain" in description + assert "adding its MX record moves mail away from any provider the domain uses today" in description + assert "allow_conflicting_provider" not in description + + +def test_setup_link_says_to_open_the_url_and_to_ask_before_replacing_a_provider(): + description = by_name("get_domain_setup_link").description + assert "open url in a browser" in description + assert "When get_domain lists an MX record, the link replaces the domain's current MX records" in description + assert "the check can miss it" in description + assert "confirm with the user that the domain has no other mail provider before opening such a link" in description + assert "When supported is false, add the records from get_domain by hand" in description + assert "verify_domain" in description + + +def test_verify_domain_says_it_returns_before_the_check_is_done(): + description = by_name("verify_domain").description + assert "Returns at once" in description + assert "call get_domain to follow it until status is VERIFIED" in description + + +def test_verify_domain_maps_each_kind_of_reason_to_the_next_step_and_says_when_to_stop(): + description = by_name("verify_domain").description + assert "a dns_records_* reason means a record is missing or wrong at the DNS provider" in description + assert "ses_*_pending and ses_*_temporary_failure clear on their own" in description + assert "ses_*_failed and ses_*_not_started need verify_domain again" in description + assert "if nothing changes after several checks, stop and tell the user the reason" in description + + +# -------------------------------------------------------------------------- +# params +# -------------------------------------------------------------------------- + + +def test_get_domain_params_need_the_domain_id(): + assert GetDomainParams(domain_id="example.com").domain_id == "example.com" + with pytest.raises(ValueError): + GetDomainParams() + + +def test_create_domain_params_need_only_the_domain(): + schema = CreateDomainParams.model_json_schema() + assert schema["required"] == ["domain"] + assert set(schema["properties"]) == { + "domain", + "feedback_enabled", + "subdomains_enabled", + "tracking_enabled", + } + with pytest.raises(ValueError): + CreateDomainParams() + + +def test_create_domain_params_state_each_default(): + properties = CreateDomainParams.model_json_schema()["properties"] + assert "Default true" in properties["feedback_enabled"]["description"] + assert "Default false" in properties["subdomains_enabled"]["description"] + assert "Default false" in properties["tracking_enabled"]["description"] + + +# -------------------------------------------------------------------------- +# the requests the SDK sends +# -------------------------------------------------------------------------- + + +def test_list_domains_pages_with_limit_and_token(): + api = FakeApi() + result = functions.list_domains(sdk(api), {"limit": 5, "page_token": "tok"}) + + assert api.requests == [("GET", "/v0/domains", {"limit": "5", "page_token": "tok"}, None)] + assert result.count == 1 + assert result.domains[0].domain == "example.com" + + +def test_get_domain_reads_that_domain(): + api = FakeApi() + result = functions.get_domain(sdk(api), {"domain_id": "example.com"}) + + assert api.requests == [("GET", "/v0/domains/example.com", {}, None)] + assert result.status == "NOT_STARTED" + assert result.records[0].status == "MISSING" + + +def test_create_domain_sends_only_the_domain_when_no_flag_is_set(): + api = FakeApi() + functions.create_domain(sdk(api), {"domain": "example.com"}) + + assert api.requests == [("POST", "/v0/domains", {}, {"domain": "example.com"})] + + +def test_create_domain_sends_every_flag_set_false_included(): + api = FakeApi() + functions.create_domain( + sdk(api), + { + "domain": "example.com", + "feedback_enabled": False, + "subdomains_enabled": True, + "tracking_enabled": False, + }, + ) + + assert api.requests[0][3] == { + "domain": "example.com", + "feedback_enabled": False, + "subdomains_enabled": True, + "tracking_enabled": False, + } + + +def test_create_domain_never_sends_allow_conflicting_provider(): + api = FakeApi() + functions.create_domain(sdk(api), {"domain": "example.com", "allow_conflicting_provider": True, "pod_id": "pod_2"}) + + assert api.requests[0][3] == {"domain": "example.com"} + + +def test_create_domain_without_a_domain_makes_no_request(): + api = FakeApi() + with pytest.raises(ValueError): + functions.create_domain(sdk(api), {"feedback_enabled": True}) + assert api.requests == [] + + +def test_get_domain_setup_link_returns_the_signed_url_unchanged(): + api = FakeApi() + result = functions.get_domain_setup_link(sdk(api), {"domain_id": "example.com"}) + + assert api.requests == [("GET", "/v0/domains/example.com/setup-link", {}, None)] + assert result.supported is True + assert result.provider_name == "Cloudflare" + assert result.url == SIGNED_URL + + +def test_verify_domain_posts_to_verify_and_returns_none(): + api = FakeApi() + result = functions.verify_domain(sdk(api), {"domain_id": "example.com"}) + + assert api.requests == [("POST", "/v0/domains/example.com/verify", {}, None)] + assert result is None + + +# -------------------------------------------------------------------------- +# adapters +# -------------------------------------------------------------------------- + + +def openai_tool(api, name): + from agentmail_toolkit.openai import AgentMailToolkit + + return AgentMailToolkit(client=sdk(api))._tools[name] + + +def test_openai_create_domain_with_every_key_present_as_null(): + """The OpenAI adapter's strict schema makes the model send every key, with null for + the flags it leaves unset. The API reads a null in a create body as unset, so no + flag may arrive with a value the model did not choose.""" + api = FakeApi() + tool = openai_tool(api, "create_domain") + arguments = { + "domain": "example.com", + "feedback_enabled": None, + "subdomains_enabled": None, + "tracking_enabled": None, + } + + result = asyncio.run(tool.on_invoke_tool(MagicMock(), json.dumps(arguments))) + + sent = api.requests[0][3] + assert {key: value for key, value in sent.items() if value is not None} == {"domain": "example.com"} + assert json.loads(result)["status"] == "NOT_STARTED" + + +def test_openai_create_domain_schema_is_strict_so_every_key_arrives(): + tool = openai_tool(FakeApi(), "create_domain") + assert tool.strict_json_schema is True + assert tool.params_json_schema["additionalProperties"] is False + assert set(tool.params_json_schema["required"]) == { + "domain", + "feedback_enabled", + "subdomains_enabled", + "tracking_enabled", + } + + +def test_openai_setup_link_result_carries_the_url(): + tool = openai_tool(FakeApi(), "get_domain_setup_link") + result = json.loads(asyncio.run(tool.on_invoke_tool(MagicMock(), json.dumps({"domain_id": "example.com"})))) + + assert result["supported"] is True + assert result["url"] == SIGNED_URL + + +def test_openai_verify_domain_returns_ok(): + tool = openai_tool(FakeApi(), "verify_domain") + assert asyncio.run(tool.on_invoke_tool(MagicMock(), json.dumps({"domain_id": "example.com"}))) == "OK" + + +def test_openai_create_domain_conflict_raises_with_the_provider_named(): + tool = openai_tool(FakeApi(error=CONFLICT_422), "create_domain") + + with pytest.raises(RuntimeError, match="Google Workspace") as raised: + asyncio.run(tool.on_invoke_tool(MagicMock(), json.dumps({"domain": "example.com"}))) + assert "HTTP 422" in str(raised.value) + + +def test_langchain_create_domain_passes_the_flags_through(): + from agentmail_toolkit.langchain import AgentMailToolkit + + api = FakeApi() + tool = AgentMailToolkit(client=sdk(api))._tools["create_domain"] + tool.invoke({"domain": "example.com", "subdomains_enabled": True}) + + # Only what the model set reaches the API with a value; nothing else may carry one. + sent = api.requests[0][3] + assert {key: value for key, value in sent.items() if value is not None} == { + "domain": "example.com", + "subdomains_enabled": True, + } + + +def test_langchain_verify_domain_refusal_is_an_error_message(): + from agentmail_toolkit.langchain import AgentMailToolkit + + api = FakeApi(error=(403, {"name": "ForbiddenError", "message": "Forbidden"})) + tool = AgentMailToolkit(client=sdk(api))._tools["verify_domain"] + + message = tool.run(tool_input={"domain_id": "example.com"}, tool_call_id="call_1") + + assert message.status == "error" + assert "HTTP 403" in message.content + with pytest.raises(LangchainToolException): + tool.func(domain_id="example.com") + + +def _livekit_context(): + context = MagicMock() + context.session.generate_reply = AsyncMock() + return context + + +def test_livekit_verify_domain_returns_ok(): + from agentmail_toolkit.livekit import AgentMailToolkit + + tool = AgentMailToolkit(client=sdk(FakeApi()))._tools["verify_domain"] + assert asyncio.run(tool({"domain_id": "example.com"}, _livekit_context())) == "OK" + + +def test_livekit_create_domain_drops_an_override_the_model_was_not_offered(): + from agentmail_toolkit.livekit import AgentMailToolkit + + api = FakeApi() + tool = AgentMailToolkit(client=sdk(api))._tools["create_domain"] + asyncio.run(tool({"domain": "example.com", "allow_conflicting_provider": True}, _livekit_context())) + + assert api.requests[0][3] == {"domain": "example.com"} + + +def test_livekit_get_domain_not_found_raises_tool_error(): + from agentmail_toolkit.livekit import AgentMailToolkit + + api = FakeApi(error=(404, {"name": "NotFoundError", "message": "Domain not found"})) + tool = AgentMailToolkit(client=sdk(api))._tools["get_domain"] + + with pytest.raises(LivekitToolError, match="Domain not found"): + asyncio.run(tool({"domain_id": "nobody.example"}, _livekit_context())) + + +# -------------------------------------------------------------------------- +# start to finish +# -------------------------------------------------------------------------- + + +def test_create_link_verify_then_get_until_verified(): + from agentmail_toolkit.openai import AgentMailToolkit + + api = FakeApi() + toolkit = AgentMailToolkit(client=sdk(api))._tools + + def call(name, arguments): + return asyncio.run(toolkit[name].on_invoke_tool(MagicMock(), json.dumps(arguments))) + + created = json.loads(call("create_domain", {"domain": "example.com"})) + domain_id = created["domain_id"] + assert created["status"] == "NOT_STARTED" + + link = json.loads(call("get_domain_setup_link", {"domain_id": domain_id})) + assert link["url"] == SIGNED_URL + + # Here the agent opens link["url"] and the domain owner approves at the provider. + + assert call("verify_domain", {"domain_id": domain_id}) == "OK" + statuses = [json.loads(call("get_domain", {"domain_id": domain_id}))["status"] for _ in range(2)] + + assert statuses == ["VERIFYING", "VERIFIED"] + assert [(method, path) for method, path, _, _ in api.requests] == [ + ("POST", "/v0/domains"), + ("GET", "/v0/domains/example.com/setup-link"), + ("POST", "/v0/domains/example.com/verify"), + ("GET", "/v0/domains/example.com"), + ("GET", "/v0/domains/example.com"), + ]