# Directory Structure
```
├── .editorconfig
├── .github
│ ├── actions
│ │ └── setup-python-env
│ │ └── action.yml
│ └── workflows
│ ├── main.yml
│ ├── on-release-main.yml
│ └── validate-codecov-config.yml
├── .gitignore
├── .pre-commit-config.yaml
├── .python-version
├── .vscode
│ └── settings.json
├── codecov.yaml
├── CONTRIBUTING.md
├── dev
│ ├── claude_desktop_config.json
│ └── install_claude_desktop.py
├── Dockerfile
├── docs
│ ├── index.md
│ └── modules.md
├── LICENSE
├── Makefile
├── mcp_email_server
│ ├── __init__.py
│ ├── app.py
│ ├── cli.py
│ ├── config.py
│ ├── emails
│ │ ├── __init__.py
│ │ ├── classic.py
│ │ ├── dispatcher.py
│ │ ├── models.py
│ │ └── provider
│ │ └── __init__.py
│ ├── log.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── claude_desktop_config.json
│ │ └── installer.py
│ └── ui.py
├── mkdocs.yml
├── pyproject.toml
├── pytest.ini
├── README.md
├── smithery.yaml
├── tests
│ ├── conftest.py
│ ├── test_classic_handler.py
│ ├── test_config.py
│ ├── test_dispatcher.py
│ ├── test_email_client.py
│ ├── test_env_config_coverage.py
│ ├── test_mcp_tools.py
│ └── test_models.py
├── tox.ini
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.12
2 |
```
--------------------------------------------------------------------------------
/.editorconfig:
--------------------------------------------------------------------------------
```
1 | max_line_length = 120
2 |
3 | [*.json]
4 | indent_style = space
5 | indent_size = 4
6 |
```
--------------------------------------------------------------------------------
/.pre-commit-config.yaml:
--------------------------------------------------------------------------------
```yaml
1 | repos:
2 | - repo: https://github.com/pre-commit/pre-commit-hooks
3 | rev: "v6.0.0"
4 | hooks:
5 | - id: check-case-conflict
6 | - id: check-merge-conflict
7 | - id: check-toml
8 | - id: check-yaml
9 | - id: end-of-file-fixer
10 | - id: trailing-whitespace
11 |
12 | - repo: https://github.com/astral-sh/ruff-pre-commit
13 | rev: "v0.14.0"
14 | hooks:
15 | - id: ruff
16 | args: [--exit-non-zero-on-fix]
17 | - id: ruff-format
18 |
19 | - repo: https://github.com/pre-commit/mirrors-prettier
20 | rev: "v4.0.0-alpha.8"
21 | hooks:
22 | - id: prettier
23 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | docs/source
2 |
3 | # From https://raw.githubusercontent.com/github/gitignore/main/Python.gitignore
4 |
5 | # Byte-compiled / optimized / DLL files
6 | __pycache__/
7 | *.py[cod]
8 | *$py.class
9 |
10 | # C extensions
11 | *.so
12 |
13 | # Distribution / packaging
14 | .Python
15 | build/
16 | develop-eggs/
17 | dist/
18 | downloads/
19 | eggs/
20 | .eggs/
21 | lib/
22 | lib64/
23 | parts/
24 | sdist/
25 | var/
26 | wheels/
27 | share/python-wheels/
28 | *.egg-info/
29 | .installed.cfg
30 | *.egg
31 | MANIFEST
32 |
33 | # PyInstaller
34 | # Usually these files are written by a python script from a template
35 | # before PyInstaller builds the exe, so as to inject date/other infos into it.
36 | *.manifest
37 | *.spec
38 |
39 | # Installer logs
40 | pip-log.txt
41 | pip-delete-this-directory.txt
42 |
43 | # Unit test / coverage reports
44 | htmlcov/
45 | .tox/
46 | .nox/
47 | .coverage
48 | .coverage.*
49 | .cache
50 | nosetests.xml
51 | coverage.xml
52 | *.cover
53 | *.py,cover
54 | .hypothesis/
55 | .pytest_cache/
56 | cover/
57 |
58 | # Translations
59 | *.mo
60 | *.pot
61 |
62 | # Django stuff:
63 | *.log
64 | local_settings.py
65 | db.sqlite3
66 | db.sqlite3-journal
67 |
68 | # Flask stuff:
69 | instance/
70 | .webassets-cache
71 |
72 | # Scrapy stuff:
73 | .scrapy
74 |
75 | # Sphinx documentation
76 | docs/_build/
77 |
78 | # PyBuilder
79 | .pybuilder/
80 | target/
81 |
82 | # Jupyter Notebook
83 | .ipynb_checkpoints
84 |
85 | # IPython
86 | profile_default/
87 | ipython_config.py
88 |
89 | # PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
90 | __pypackages__/
91 |
92 | # Celery stuff
93 | celerybeat-schedule
94 | celerybeat.pid
95 |
96 | # SageMath parsed files
97 | *.sage.py
98 |
99 | # Environments
100 | .env
101 | .venv
102 | env/
103 | venv/
104 | ENV/
105 | env.bak/
106 | venv.bak/
107 |
108 | # Spyder project settings
109 | .spyderproject
110 | .spyproject
111 |
112 | # Rope project settings
113 | .ropeproject
114 |
115 | # mkdocs documentation
116 | /site
117 |
118 | # mypy
119 | .mypy_cache/
120 | .dmypy.json
121 | dmypy.json
122 |
123 | # Pyre type checker
124 | .pyre/
125 |
126 | # pytype static type analyzer
127 | .pytype/
128 |
129 | # Cython debug symbols
130 | cython_debug/
131 |
132 | # Vscode config files
133 | # .vscode/
134 |
135 | # PyCharm
136 | # JetBrains specific template is maintained in a separate JetBrains.gitignore that can
137 | # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
138 | # and can be added to the global gitignore or merged into this file. For a more nuclear
139 | # option (not recommended) you can uncomment the following to ignore the entire idea folder.
140 | #.idea/
141 |
142 | tests/config.toml
143 | local/
144 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # mcp-email-server
2 |
3 | [](https://img.shields.io/github/v/release/ai-zerolab/mcp-email-server)
4 | [](https://github.com/ai-zerolab/mcp-email-server/actions/workflows/main.yml?query=branch%3Amain)
5 | [](https://codecov.io/gh/ai-zerolab/mcp-email-server)
6 | [](https://img.shields.io/github/commit-activity/m/ai-zerolab/mcp-email-server)
7 | [](https://img.shields.io/github/license/ai-zerolab/mcp-email-server)
8 | [](https://smithery.ai/server/@ai-zerolab/mcp-email-server)
9 |
10 | IMAP and SMTP via MCP Server
11 |
12 | - **Github repository**: <https://github.com/ai-zerolab/mcp-email-server/>
13 | - **Documentation** <https://ai-zerolab.github.io/mcp-email-server/>
14 |
15 | ## Installation
16 |
17 | ### Manual Installation
18 |
19 | We recommend using [uv](https://github.com/astral-sh/uv) to manage your environment.
20 |
21 | Try `uvx mcp-email-server@latest ui` to config, and use following configuration for mcp client:
22 |
23 | ```json
24 | {
25 | "mcpServers": {
26 | "zerolib-email": {
27 | "command": "uvx",
28 | "args": ["mcp-email-server@latest", "stdio"]
29 | }
30 | }
31 | }
32 | ```
33 |
34 | This package is available on PyPI, so you can install it using `pip install mcp-email-server`
35 |
36 | After that, configure your email server using the ui: `mcp-email-server ui`
37 |
38 | ### Environment Variable Configuration
39 |
40 | You can also configure the email server using environment variables, which is particularly useful for CI/CD environments like Jenkins. zerolib-email supports both UI configuration (via TOML file) and environment variables, with environment variables taking precedence.
41 |
42 | ```json
43 | {
44 | "mcpServers": {
45 | "zerolib-email": {
46 | "command": "uvx",
47 | "args": ["mcp-email-server@latest", "stdio"],
48 | "env": {
49 | "MCP_EMAIL_SERVER_ACCOUNT_NAME": "work",
50 | "MCP_EMAIL_SERVER_FULL_NAME": "John Doe",
51 | "MCP_EMAIL_SERVER_EMAIL_ADDRESS": "[email protected]",
52 | "MCP_EMAIL_SERVER_USER_NAME": "[email protected]",
53 | "MCP_EMAIL_SERVER_PASSWORD": "your_password",
54 | "MCP_EMAIL_SERVER_IMAP_HOST": "imap.gmail.com",
55 | "MCP_EMAIL_SERVER_IMAP_PORT": "993",
56 | "MCP_EMAIL_SERVER_SMTP_HOST": "smtp.gmail.com",
57 | "MCP_EMAIL_SERVER_SMTP_PORT": "465"
58 | }
59 | }
60 | }
61 | }
62 | ```
63 |
64 | #### Available Environment Variables
65 |
66 | | Variable | Description | Default | Required |
67 | | --------------------------------- | ------------------ | ------------- | -------- |
68 | | `MCP_EMAIL_SERVER_ACCOUNT_NAME` | Account identifier | `"default"` | No |
69 | | `MCP_EMAIL_SERVER_FULL_NAME` | Display name | Email prefix | No |
70 | | `MCP_EMAIL_SERVER_EMAIL_ADDRESS` | Email address | - | Yes |
71 | | `MCP_EMAIL_SERVER_USER_NAME` | Login username | Same as email | No |
72 | | `MCP_EMAIL_SERVER_PASSWORD` | Email password | - | Yes |
73 | | `MCP_EMAIL_SERVER_IMAP_HOST` | IMAP server host | - | Yes |
74 | | `MCP_EMAIL_SERVER_IMAP_PORT` | IMAP server port | `993` | No |
75 | | `MCP_EMAIL_SERVER_IMAP_SSL` | Enable IMAP SSL | `true` | No |
76 | | `MCP_EMAIL_SERVER_SMTP_HOST` | SMTP server host | - | Yes |
77 | | `MCP_EMAIL_SERVER_SMTP_PORT` | SMTP server port | `465` | No |
78 | | `MCP_EMAIL_SERVER_SMTP_SSL` | Enable SMTP SSL | `true` | No |
79 | | `MCP_EMAIL_SERVER_SMTP_START_SSL` | Enable STARTTLS | `false` | No |
80 |
81 | For separate IMAP/SMTP credentials, you can also use:
82 |
83 | - `MCP_EMAIL_SERVER_IMAP_USER_NAME` / `MCP_EMAIL_SERVER_IMAP_PASSWORD`
84 | - `MCP_EMAIL_SERVER_SMTP_USER_NAME` / `MCP_EMAIL_SERVER_SMTP_PASSWORD`
85 |
86 | Then you can try it in [Claude Desktop](https://claude.ai/download). If you want to intergrate it with other mcp client, run `$which mcp-email-server` for the path and configure it in your client like:
87 |
88 | ```json
89 | {
90 | "mcpServers": {
91 | "zerolib-email": {
92 | "command": "{{ ENTRYPOINT }}",
93 | "args": ["stdio"]
94 | }
95 | }
96 | }
97 | ```
98 |
99 | If `docker` is avaliable, you can try use docker image, but you may need to config it in your client using `tools` via `MCP`. The default config path is `~/.config/zerolib/mcp_email_server/config.toml`
100 |
101 | ```json
102 | {
103 | "mcpServers": {
104 | "zerolib-email": {
105 | "command": "docker",
106 | "args": ["run", "-it", "ghcr.io/ai-zerolab/mcp-email-server:latest"]
107 | }
108 | }
109 | }
110 | ```
111 |
112 | ### Installing via Smithery
113 |
114 | To install Email Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@ai-zerolab/mcp-email-server):
115 |
116 | ```bash
117 | npx -y @smithery/cli install @ai-zerolab/mcp-email-server --client claude
118 | ```
119 |
120 | ## Development
121 |
122 | This project is managed using [uv](https://github.com/ai-zerolab/uv).
123 |
124 | Try `make install` to install the virtual environment and install the pre-commit hooks.
125 |
126 | Use `uv run mcp-email-server` for local development.
127 |
128 | ## Releasing a new version
129 |
130 | - Create an API Token on [PyPI](https://pypi.org/).
131 | - Add the API Token to your projects secrets with the name `PYPI_TOKEN` by visiting [this page](https://github.com/ai-zerolab/mcp-email-server/settings/secrets/actions/new).
132 | - Create a [new release](https://github.com/ai-zerolab/mcp-email-server/releases/new) on Github.
133 | - Create a new tag in the form `*.*.*`.
134 |
135 | For more details, see [here](https://fpgmaas.github.io/cookiecutter-uv/features/cicd/#how-to-trigger-a-release).
136 |
```
--------------------------------------------------------------------------------
/CONTRIBUTING.md:
--------------------------------------------------------------------------------
```markdown
1 | # Contributing to `mcp-email-server`
2 |
3 | Contributions are welcome, and they are greatly appreciated!
4 | Every little bit helps, and credit will always be given.
5 |
6 | You can contribute in many ways:
7 |
8 | # Types of Contributions
9 |
10 | ## Report Bugs
11 |
12 | Report bugs at https://github.com/ai-zerolab/mcp-email-server/issues
13 |
14 | If you are reporting a bug, please include:
15 |
16 | - Your operating system name and version.
17 | - Any details about your local setup that might be helpful in troubleshooting.
18 | - Detailed steps to reproduce the bug.
19 |
20 | ## Fix Bugs
21 |
22 | Look through the GitHub issues for bugs.
23 | Anything tagged with "bug" and "help wanted" is open to whoever wants to implement a fix for it.
24 |
25 | ## Implement Features
26 |
27 | Look through the GitHub issues for features.
28 | Anything tagged with "enhancement" and "help wanted" is open to whoever wants to implement it.
29 |
30 | ## Write Documentation
31 |
32 | mcp-email-server could always use more documentation, whether as part of the official docs, in docstrings, or even on the web in blog posts, articles, and such.
33 |
34 | ## Submit Feedback
35 |
36 | The best way to send feedback is to file an issue at https://github.com/ai-zerolab/mcp-email-server/issues.
37 |
38 | If you are proposing a new feature:
39 |
40 | - Explain in detail how it would work.
41 | - Keep the scope as narrow as possible, to make it easier to implement.
42 | - Remember that this is a volunteer-driven project, and that contributions
43 | are welcome :)
44 |
45 | # Get Started!
46 |
47 | Ready to contribute? Here's how to set up `mcp-email-server` for local development.
48 | Please note this documentation assumes you already have `uv` and `Git` installed and ready to go.
49 |
50 | 1. Fork the `mcp-email-server` repo on GitHub.
51 |
52 | 2. Clone your fork locally:
53 |
54 | ```bash
55 | cd <directory_in_which_repo_should_be_created>
56 | git clone [email protected]:YOUR_NAME/mcp-email-server.git
57 | ```
58 |
59 | 3. Now we need to install the environment. Navigate into the directory
60 |
61 | ```bash
62 | cd mcp-email-server
63 | ```
64 |
65 | Then, install and activate the environment with:
66 |
67 | ```bash
68 | uv sync
69 | ```
70 |
71 | 4. Install pre-commit to run linters/formatters at commit time:
72 |
73 | ```bash
74 | uv run pre-commit install
75 | ```
76 |
77 | 5. Create a branch for local development:
78 |
79 | ```bash
80 | git checkout -b name-of-your-bugfix-or-feature
81 | ```
82 |
83 | Now you can make your changes locally.
84 |
85 | 6. Don't forget to add test cases for your added functionality to the `tests` directory.
86 |
87 | 7. When you're done making changes, check that your changes pass the formatting tests.
88 |
89 | ```bash
90 | make check
91 | ```
92 |
93 | Now, validate that all unit tests are passing:
94 |
95 | ```bash
96 | make test
97 | ```
98 |
99 | 9. Before raising a pull request you should also run tox.
100 | This will run the tests across different versions of Python:
101 |
102 | ```bash
103 | tox
104 | ```
105 |
106 | This requires you to have multiple versions of python installed.
107 | This step is also triggered in the CI/CD pipeline, so you could also choose to skip this step locally.
108 |
109 | 10. Commit your changes and push your branch to GitHub:
110 |
111 | ```bash
112 | git add .
113 | git commit -m "Your detailed description of your changes."
114 | git push origin name-of-your-bugfix-or-feature
115 | ```
116 |
117 | 11. Submit a pull request through the GitHub website.
118 |
119 | # Pull Request Guidelines
120 |
121 | Before you submit a pull request, check that it meets these guidelines:
122 |
123 | 1. The pull request should include tests.
124 |
125 | 2. If the pull request adds functionality, the docs should be updated.
126 | Put your new functionality into a function with a docstring, and add the feature to the list in `README.md`.
127 |
```
--------------------------------------------------------------------------------
/mcp_email_server/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/mcp_email_server/emails/provider/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/mcp_email_server/tools/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/docs/modules.md:
--------------------------------------------------------------------------------
```markdown
1 | ::: mcp_email_server.app
2 |
```
--------------------------------------------------------------------------------
/.vscode/settings.json:
--------------------------------------------------------------------------------
```json
1 | {
2 | "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python"
3 | }
4 |
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
1 | # pytest.ini
2 | [pytest]
3 | asyncio_mode = auto
4 | # asyncio_default_fixture_loop_scope =
5 |
```
--------------------------------------------------------------------------------
/mcp_email_server/tools/claude_desktop_config.json:
--------------------------------------------------------------------------------
```json
1 | {
2 | "mcpServers": {
3 | "zerolib-email": {
4 | "command": "{{ ENTRYPOINT }}",
5 | "args": ["stdio"]
6 | }
7 | }
8 | }
9 |
```
--------------------------------------------------------------------------------
/dev/claude_desktop_config.json:
--------------------------------------------------------------------------------
```json
1 | {
2 | "mcpServers": {
3 | "email-dev": {
4 | "command": "{{ PWD }}/.venv/bin/python {{ PWD }}/mcp_email_server/cli.py",
5 | "args": ["stdio"]
6 | }
7 | }
8 | }
9 |
```
--------------------------------------------------------------------------------
/codecov.yaml:
--------------------------------------------------------------------------------
```yaml
1 | coverage:
2 | range: 70..100
3 | round: down
4 | precision: 1
5 | status:
6 | project:
7 | default:
8 | target: 90%
9 | threshold: 0.5%
10 | codecov:
11 | token: f927bff4-d404-4986-8c11-624eadda8431
12 |
```
--------------------------------------------------------------------------------
/mcp_email_server/log.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 |
3 | USER_DEFINED_LOG_LEVEL = os.getenv("MCP_EMAIL_SERVER_LOG_LEVEL", "INFO")
4 |
5 | os.environ["LOGURU_LEVEL"] = USER_DEFINED_LOG_LEVEL
6 |
7 | from loguru import logger # noqa: E402
8 |
9 | __all__ = ["logger"]
10 |
```
--------------------------------------------------------------------------------
/.github/workflows/validate-codecov-config.yml:
--------------------------------------------------------------------------------
```yaml
1 | name: validate-codecov-config
2 |
3 | on:
4 | pull_request:
5 | paths: [codecov.yaml]
6 | push:
7 | branches: [main]
8 |
9 | jobs:
10 | validate-codecov-config:
11 | runs-on: ubuntu-22.04
12 | steps:
13 | - uses: actions/checkout@v4
14 | - name: Validate codecov configuration
15 | run: curl -sSL --fail-with-body --data-binary @codecov.yaml https://codecov.io/validate
16 |
```
--------------------------------------------------------------------------------
/tox.ini:
--------------------------------------------------------------------------------
```
1 | [tox]
2 | skipsdist = true
3 | envlist = py310, py311, py312, py313, py314
4 |
5 | [gh-actions]
6 | python =
7 | 3.10: py310
8 | 3.11: py311
9 | 3.12: py312
10 | 3.13: py313
11 | ; 3.14: py314
12 |
13 | [testenv]
14 | passenv = PYTHON_VERSION
15 | allowlist_externals = uv
16 | commands =
17 | uv sync --python {envpython}
18 | uv run python -m pytest --doctest-modules tests --cov --cov-config=pyproject.toml --cov-report=xml
19 |
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
1 | # Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
2 |
3 | startCommand:
4 | type: stdio
5 | configSchema:
6 | # JSON Schema defining the configuration options for the MCP.
7 | type: object
8 | properties: {}
9 | commandFunction:
10 | # A JS function that produces the CLI command based on the given config to start the MCP on stdio.
11 | |-
12 | (config) => ({
13 | command: 'uv',
14 | args: ['run', 'mcp-email-server@latest', 'stdio'],
15 | env: {}
16 | })
17 | exampleConfig: {}
18 |
```
--------------------------------------------------------------------------------
/mcp_email_server/cli.py:
--------------------------------------------------------------------------------
```python
1 | import typer
2 |
3 | from mcp_email_server.app import mcp
4 | from mcp_email_server.config import delete_settings
5 |
6 | app = typer.Typer()
7 |
8 |
9 | @app.command()
10 | def stdio():
11 | mcp.run(transport="stdio")
12 |
13 |
14 | @app.command()
15 | def sse(
16 | host: str = "localhost",
17 | port: int = 9557,
18 | ):
19 | mcp.settings.host = host
20 | mcp.settings.port = port
21 | mcp.run(transport="sse")
22 |
23 |
24 | @app.command()
25 | def ui():
26 | from mcp_email_server.ui import main as ui_main
27 |
28 | ui_main()
29 |
30 |
31 | @app.command()
32 | def reset():
33 | delete_settings()
34 | typer.echo("✅ Config reset")
35 |
36 |
37 | if __name__ == "__main__":
38 | app(["stdio"])
39 |
```
--------------------------------------------------------------------------------
/mcp_email_server/emails/dispatcher.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | from typing import TYPE_CHECKING
4 |
5 | from mcp_email_server.config import EmailSettings, ProviderSettings, get_settings
6 | from mcp_email_server.emails.classic import ClassicEmailHandler
7 |
8 | if TYPE_CHECKING:
9 | from mcp_email_server.emails import EmailHandler
10 |
11 |
12 | def dispatch_handler(account_name: str) -> EmailHandler:
13 | settings = get_settings()
14 | account = settings.get_account(account_name)
15 | if isinstance(account, ProviderSettings):
16 | raise NotImplementedError
17 | if isinstance(account, EmailSettings):
18 | return ClassicEmailHandler(account)
19 |
20 | raise ValueError(f"Account {account_name} not found, available accounts: {settings.get_accounts()}")
21 |
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | # Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
2 | # Install uv
3 | FROM python:3.12-slim
4 |
5 | # Install tini
6 | RUN apt-get update && \
7 | apt-get install -y --no-install-recommends tini && \
8 | rm -rf /var/lib/apt/lists/*
9 |
10 | COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
11 |
12 | # Change the working directory to the `app` directory
13 | WORKDIR /app
14 |
15 | # Copy the lockfile and `pyproject.toml` into the image
16 | COPY uv.lock /app/uv.lock
17 | COPY pyproject.toml /app/pyproject.toml
18 |
19 | # Install dependencies
20 | RUN uv sync --frozen --no-install-project
21 |
22 | # Copy the project into the image
23 | COPY . /app
24 |
25 | # Sync the project
26 | RUN uv sync --frozen
27 |
28 | # Run the server
29 | ENTRYPOINT ["tini", "--", "uv", "run", "mcp-email-server"]
30 | CMD ["stdio"]
31 |
```
--------------------------------------------------------------------------------
/docs/index.md:
--------------------------------------------------------------------------------
```markdown
1 | # mcp-email-server
2 |
3 | [](https://img.shields.io/github/v/release/ai-zerolab/mcp-email-server)
4 | [](https://github.com/ai-zerolab/mcp-email-server/actions/workflows/main.yml?query=branch%3Amain)
5 | [](https://img.shields.io/github/commit-activity/m/ai-zerolab/mcp-email-server)
6 | [](https://img.shields.io/github/license/ai-zerolab/mcp-email-server)
7 |
8 | IMAP and SMTP via MCP Server
9 |
```
--------------------------------------------------------------------------------
/.github/actions/setup-python-env/action.yml:
--------------------------------------------------------------------------------
```yaml
1 | name: "Setup Python Environment"
2 | description: "Set up Python environment for the given Python version"
3 |
4 | inputs:
5 | python-version:
6 | description: "Python version to use"
7 | required: true
8 | default: "3.12"
9 | uv-version:
10 | description: "uv version to use"
11 | required: true
12 | default: "0.6.2"
13 |
14 | runs:
15 | using: "composite"
16 | steps:
17 | - uses: actions/setup-python@v5
18 | with:
19 | python-version: ${{ inputs.python-version }}
20 |
21 | - name: Install uv
22 | uses: astral-sh/setup-uv@v2
23 | with:
24 | version: ${{ inputs.uv-version }}
25 | enable-cache: "true"
26 | cache-suffix: ${{ matrix.python-version }}
27 |
28 | - name: Install Python dependencies
29 | run: uv sync --frozen --python=${{ matrix.python-version }}
30 | shell: bash
31 |
```
--------------------------------------------------------------------------------
/mcp_email_server/emails/__init__.py:
--------------------------------------------------------------------------------
```python
1 | import abc
2 | from datetime import datetime
3 | from typing import TYPE_CHECKING
4 |
5 | if TYPE_CHECKING:
6 | from mcp_email_server.emails.models import EmailContentBatchResponse, EmailMetadataPageResponse
7 |
8 |
9 | class EmailHandler(abc.ABC):
10 | @abc.abstractmethod
11 | async def get_emails_metadata(
12 | self,
13 | page: int = 1,
14 | page_size: int = 10,
15 | before: datetime | None = None,
16 | since: datetime | None = None,
17 | subject: str | None = None,
18 | from_address: str | None = None,
19 | to_address: str | None = None,
20 | order: str = "desc",
21 | ) -> "EmailMetadataPageResponse":
22 | """
23 | Get email metadata only (without body content) for better performance
24 | """
25 |
26 | @abc.abstractmethod
27 | async def get_emails_content(self, email_ids: list[str]) -> "EmailContentBatchResponse":
28 | """
29 | Get full content (including body) of multiple emails by their email IDs (IMAP UIDs)
30 | """
31 |
32 | @abc.abstractmethod
33 | async def send_email(
34 | self,
35 | recipients: list[str],
36 | subject: str,
37 | body: str,
38 | cc: list[str] | None = None,
39 | bcc: list[str] | None = None,
40 | html: bool = False,
41 | ) -> None:
42 | """
43 | Send email
44 | """
45 |
```
--------------------------------------------------------------------------------
/mkdocs.yml:
--------------------------------------------------------------------------------
```yaml
1 | site_name: mcp-email-server
2 | repo_url: https://github.com/ai-zerolab/mcp-email-server
3 | site_url: https://ai-zerolab.github.io/mcp-email-server
4 | site_description: IMAP and SMTP via MCP Server
5 | site_author: ai-zerolab
6 | edit_uri: edit/main/docs/
7 | repo_name: ai-zerolab/mcp-email-server
8 | copyright: Maintained by <a href="https://ai-zerolab.com">ai-zerolab</a>.
9 |
10 | nav:
11 | - Home: index.md
12 | - Modules: modules.md
13 | plugins:
14 | - search
15 | - mkdocstrings:
16 | handlers:
17 | python:
18 | paths: ["mcp_email_server"]
19 | theme:
20 | name: material
21 | feature:
22 | tabs: true
23 | palette:
24 | - media: "(prefers-color-scheme: light)"
25 | scheme: default
26 | primary: white
27 | accent: deep orange
28 | toggle:
29 | icon: material/brightness-7
30 | name: Switch to dark mode
31 | - media: "(prefers-color-scheme: dark)"
32 | scheme: slate
33 | primary: black
34 | accent: deep orange
35 | toggle:
36 | icon: material/brightness-4
37 | name: Switch to light mode
38 | icon:
39 | repo: fontawesome/brands/github
40 |
41 | extra:
42 | social:
43 | - icon: fontawesome/brands/github
44 | link: https://github.com/ai-zerolab/mcp-email-server
45 | - icon: fontawesome/brands/python
46 | link: https://pypi.org/project/mcp-email-server
47 |
48 | markdown_extensions:
49 | - toc:
50 | permalink: true
51 | - pymdownx.arithmatex:
52 | generic: true
53 |
```
--------------------------------------------------------------------------------
/mcp_email_server/emails/models.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import datetime
2 | from typing import Any
3 |
4 | from pydantic import BaseModel
5 |
6 |
7 | class EmailMetadata(BaseModel):
8 | """Email metadata"""
9 |
10 | email_id: str
11 | subject: str
12 | sender: str
13 | recipients: list[str] # Recipient list
14 | date: datetime
15 | attachments: list[str]
16 |
17 | @classmethod
18 | def from_email(cls, email: dict[str, Any]):
19 | return cls(
20 | email_id=email["email_id"],
21 | subject=email["subject"],
22 | sender=email["from"],
23 | recipients=email.get("to", []),
24 | date=email["date"],
25 | attachments=email["attachments"],
26 | )
27 |
28 |
29 | class EmailMetadataPageResponse(BaseModel):
30 | """Paged email metadata response"""
31 |
32 | page: int
33 | page_size: int
34 | before: datetime | None
35 | since: datetime | None
36 | subject: str | None
37 | emails: list[EmailMetadata]
38 | total: int
39 |
40 |
41 | class EmailBodyResponse(BaseModel):
42 | """Single email body response"""
43 |
44 | email_id: str # IMAP UID of this email
45 | subject: str
46 | sender: str
47 | recipients: list[str]
48 | date: datetime
49 | body: str
50 | attachments: list[str]
51 |
52 |
53 | class EmailContentBatchResponse(BaseModel):
54 | """Batch email content response for multiple emails"""
55 |
56 | emails: list[EmailBodyResponse]
57 | requested_count: int
58 | retrieved_count: int
59 | failed_ids: list[str]
60 |
```
--------------------------------------------------------------------------------
/.github/workflows/main.yml:
--------------------------------------------------------------------------------
```yaml
1 | name: Main
2 |
3 | on:
4 | push:
5 | branches:
6 | - main
7 | pull_request:
8 | types: [opened, synchronize, reopened, ready_for_review]
9 |
10 | jobs:
11 | quality:
12 | runs-on: ubuntu-latest
13 | steps:
14 | - name: Check out
15 | uses: actions/checkout@v4
16 |
17 | - uses: actions/cache@v4
18 | with:
19 | path: ~/.cache/pre-commit
20 | key: pre-commit-${{ hashFiles('.pre-commit-config.yaml') }}
21 |
22 | - name: Set up the environment
23 | uses: ./.github/actions/setup-python-env
24 |
25 | - name: Run checks
26 | run: make check
27 |
28 | tests-and-type-check:
29 | runs-on: ubuntu-latest
30 | strategy:
31 | matrix:
32 | python-version: ["3.10", "3.11", "3.12", "3.13"]
33 | fail-fast: false
34 | defaults:
35 | run:
36 | shell: bash
37 | steps:
38 | - name: Check out
39 | uses: actions/checkout@v4
40 |
41 | - name: Set up the environment
42 | uses: ./.github/actions/setup-python-env
43 | with:
44 | python-version: ${{ matrix.python-version }}
45 |
46 | - name: Run tests
47 | run: uv run python -m pytest tests --cov --cov-config=pyproject.toml --cov-report=xml
48 |
49 | - name: Upload coverage reports to Codecov with GitHub Action on Python 3.12
50 | uses: codecov/codecov-action@v4
51 | if: ${{ matrix.python-version == '3.12' }}
52 |
53 | check-docs:
54 | runs-on: ubuntu-latest
55 | steps:
56 | - name: Check out
57 | uses: actions/checkout@v4
58 |
59 | - name: Set up the environment
60 | uses: ./.github/actions/setup-python-env
61 |
62 | - name: Check if documentation can be built
63 | run: uv run mkdocs build -s
64 |
```
--------------------------------------------------------------------------------
/dev/install_claude_desktop.py:
--------------------------------------------------------------------------------
```python
1 | import json
2 | import os
3 | import platform
4 | from pathlib import Path
5 |
6 | from jinja2 import Template
7 |
8 | _HERE = Path(__file__).parent
9 | config_template = _HERE / "claude_desktop_config.json"
10 |
11 |
12 | def generate_claude_config():
13 | # Get current working directory
14 | pwd = Path.cwd().resolve().as_posix()
15 |
16 | # Read the template config
17 |
18 | template_content = config_template.read_text()
19 | rendered_content = Template(template_content).render(PWD=pwd)
20 | template_config = json.loads(rendered_content)
21 |
22 | # Determine the correct Claude config path based on the OS
23 | system = platform.system()
24 | if system == "Darwin":
25 | config_path = os.path.expanduser("~/Library/Application Support/Claude/claude_desktop_config.json")
26 | elif system == "Windows":
27 | config_path = os.path.join(os.environ["APPDATA"], "Claude", "claude_desktop_config.json")
28 | else:
29 | print("Unsupported operating system.")
30 | return
31 |
32 | # Read the existing config file or create an empty JSON object
33 | try:
34 | with open(config_path) as f:
35 | existing_config = json.load(f)
36 | except FileNotFoundError:
37 | existing_config = {}
38 |
39 | # Merge the template config into the existing config
40 | if "mcpServers" not in existing_config:
41 | existing_config["mcpServers"] = {}
42 | existing_config["mcpServers"].update(template_config["mcpServers"])
43 |
44 | # Write the merged config back to the Claude config file
45 | os.makedirs(os.path.dirname(config_path), exist_ok=True)
46 | with open(config_path, "w") as f:
47 | json.dump(existing_config, f, indent=4)
48 |
49 | print(
50 | f"""
51 | Claude Desktop configuration generated successfully.
52 |
53 | $cat {config_path}
54 | {json.dumps(existing_config, indent=4)}
55 | """
56 | )
57 |
58 |
59 | if __name__ == "__main__":
60 | generate_claude_config()
61 |
```
--------------------------------------------------------------------------------
/tests/test_config.py:
--------------------------------------------------------------------------------
```python
1 | import pytest
2 | from pydantic import ValidationError
3 |
4 | from mcp_email_server.config import (
5 | EmailServer,
6 | EmailSettings,
7 | ProviderSettings,
8 | get_settings,
9 | store_settings,
10 | )
11 |
12 |
13 | def test_config():
14 | settings = get_settings()
15 | assert settings.emails == []
16 | settings.emails.append(
17 | EmailSettings(
18 | account_name="email_test",
19 | full_name="Test User",
20 | email_address="[email protected]",
21 | incoming=EmailServer(
22 | user_name="test",
23 | password="test",
24 | host="imap.gmail.com",
25 | port=993,
26 | ssl=True,
27 | ),
28 | outgoing=EmailServer(
29 | user_name="test",
30 | password="test",
31 | host="smtp.gmail.com",
32 | port=587,
33 | ssl=True,
34 | ),
35 | )
36 | )
37 | settings.providers.append(ProviderSettings(account_name="provider_test", provider_name="test", api_key="test"))
38 | store_settings(settings)
39 | reloaded_settings = get_settings(reload=True)
40 | assert reloaded_settings == settings
41 |
42 | with pytest.raises(ValidationError):
43 | settings.add_email(
44 | EmailSettings(
45 | account_name="email_test",
46 | full_name="Test User",
47 | email_address="[email protected]",
48 | incoming=EmailServer(
49 | user_name="test",
50 | password="test",
51 | host="imap.gmail.com",
52 | port=993,
53 | ssl=True,
54 | ),
55 | outgoing=EmailServer(
56 | user_name="test",
57 | password="test",
58 | host="smtp.gmail.com",
59 | port=587,
60 | ssl=True,
61 | ),
62 | )
63 | )
64 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "mcp-email-server"
3 | version = "0.0.1"
4 | description = "IMAP and SMTP via MCP Server"
5 | authors = [{ name = "ai-zerolab", email = "[email protected]" }]
6 | readme = "README.md"
7 | keywords = ["MCP", "IMAP", "SMTP", "email"]
8 | requires-python = ">=3.10,<4.0"
9 | classifiers = [
10 | "Intended Audience :: Developers",
11 | "Programming Language :: Python",
12 | "Programming Language :: Python :: 3",
13 | "Programming Language :: Python :: 3.10",
14 | "Programming Language :: Python :: 3.11",
15 | "Programming Language :: Python :: 3.12",
16 | "Programming Language :: Python :: 3.13",
17 | "Topic :: Software Development :: Libraries :: Python Modules",
18 | ]
19 | dependencies = [
20 | "aioimaplib>=2.0.1",
21 | "aiosmtplib>=4.0.0",
22 | "gradio>=5.18.0",
23 | "jinja2>=3.1.5",
24 | "loguru>=0.7.3",
25 | "mcp[cli]>=1.3.0",
26 | "pydantic>=2.10.6",
27 | "pydantic-settings[toml]>=2.8.0",
28 | "tomli-w>=1.2.0",
29 | "typer>=0.15.1",
30 | ]
31 |
32 | [project.scripts]
33 | mcp-email-server = "mcp_email_server.cli:app"
34 |
35 | [project.urls]
36 | Homepage = "https://ai-zerolab.github.io/mcp-email-server/"
37 | Repository = "https://github.com/ai-zerolab/mcp-email-server"
38 | Documentation = "https://ai-zerolab.github.io/mcp-email-server/"
39 |
40 | [dependency-groups]
41 | dev = [
42 | "pytest>=7.2.0",
43 | "pytest-asyncio>=0.25.3",
44 | "pre-commit>=2.20.0",
45 | "tox-uv>=1.11.3",
46 | "deptry>=0.22.0",
47 | "pytest-cov>=4.0.0",
48 | "ruff>=0.9.2",
49 | "mkdocs>=1.4.2",
50 | "mkdocs-material>=8.5.10",
51 | "mkdocstrings[python]>=0.26.1",
52 | ]
53 |
54 | [build-system]
55 | requires = ["hatchling"]
56 | build-backend = "hatchling.build"
57 |
58 | [tool.setuptools]
59 | py-modules = ["mcp_email_server"]
60 |
61 | [tool.pytest.ini_options]
62 | testpaths = ["tests"]
63 |
64 | [tool.ruff]
65 | target-version = "py310"
66 | line-length = 120
67 | fix = true
68 |
69 | [tool.ruff.lint]
70 | select = [
71 | # flake8-2020
72 | "YTT",
73 | # flake8-bandit
74 | "S",
75 | # flake8-bugbear
76 | "B",
77 | # flake8-builtins
78 | "A",
79 | # flake8-comprehensions
80 | "C4",
81 | # flake8-debugger
82 | "T10",
83 | # flake8-simplify
84 | "SIM",
85 | # isort
86 | "I",
87 | # mccabe
88 | "C90",
89 | # pycodestyle
90 | "E",
91 | "W",
92 | # pyflakes
93 | "F",
94 | # pygrep-hooks
95 | "PGH",
96 | # pyupgrade
97 | "UP",
98 | # ruff
99 | "RUF",
100 | # tryceratops
101 | "TRY",
102 | ]
103 | ignore = [
104 | # LineTooLong
105 | "E501",
106 | # DoNotAssignLambda
107 | "E731",
108 | # raise-vanilla-args
109 | "TRY003",
110 | # try-consider-else
111 | "TRY300",
112 | ]
113 |
114 | [tool.ruff.lint.per-file-ignores]
115 | "tests/*" = ["S101", "S106", "SIM117"]
116 |
117 | [tool.ruff.format]
118 | preview = true
119 |
120 | [tool.coverage.report]
121 | skip_empty = true
122 |
123 | [tool.coverage.run]
124 | branch = true
125 | source = ["mcp_email_server"]
126 |
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import asyncio
4 | import os
5 | from datetime import datetime
6 | from pathlib import Path
7 | from unittest.mock import AsyncMock
8 |
9 | import pytest
10 |
11 | from mcp_email_server.config import EmailServer, EmailSettings, ProviderSettings, delete_settings
12 |
13 | _HERE = Path(__file__).resolve().parent
14 |
15 | os.environ["MCP_EMAIL_SERVER_CONFIG_PATH"] = (_HERE / "config.toml").as_posix()
16 | os.environ["MCP_EMAIL_SERVER_LOG_LEVEL"] = "DEBUG"
17 |
18 |
19 | @pytest.fixture(autouse=True)
20 | def patch_env(monkeypatch: pytest.MonkeyPatch, tmp_path: pytest.TempPathFactory):
21 | delete_settings()
22 | yield
23 |
24 |
25 | @pytest.fixture
26 | def email_server():
27 | """Fixture for a test EmailServer."""
28 | return EmailServer(
29 | user_name="test_user",
30 | password="test_password",
31 | host="test.example.com",
32 | port=993,
33 | use_ssl=True,
34 | )
35 |
36 |
37 | @pytest.fixture
38 | def email_settings():
39 | """Fixture for test EmailSettings."""
40 | return EmailSettings(
41 | account_name="test_account",
42 | full_name="Test User",
43 | email_address="[email protected]",
44 | incoming=EmailServer(
45 | user_name="test_user",
46 | password="test_password",
47 | host="imap.example.com",
48 | port=993,
49 | use_ssl=True,
50 | ),
51 | outgoing=EmailServer(
52 | user_name="test_user",
53 | password="test_password",
54 | host="smtp.example.com",
55 | port=465,
56 | use_ssl=True,
57 | ),
58 | )
59 |
60 |
61 | @pytest.fixture
62 | def provider_settings():
63 | """Fixture for test ProviderSettings."""
64 | return ProviderSettings(
65 | account_name="test_provider",
66 | provider_name="test_provider",
67 | api_key="test_api_key",
68 | )
69 |
70 |
71 | @pytest.fixture
72 | def mock_imap():
73 | """Fixture for a mocked IMAP client."""
74 | mock_imap = AsyncMock()
75 | mock_imap._client_task = asyncio.Future()
76 | mock_imap._client_task.set_result(None)
77 | mock_imap.wait_hello_from_server = AsyncMock()
78 | mock_imap.login = AsyncMock()
79 | mock_imap.select = AsyncMock()
80 | mock_imap.search = AsyncMock(return_value=(None, [b"1 2 3"]))
81 | mock_imap.fetch = AsyncMock(return_value=(None, [b"HEADER", bytearray(b"EMAIL CONTENT")]))
82 | mock_imap.logout = AsyncMock()
83 | return mock_imap
84 |
85 |
86 | @pytest.fixture
87 | def mock_smtp():
88 | """Fixture for a mocked SMTP client."""
89 | mock_smtp = AsyncMock()
90 | mock_smtp.__aenter__.return_value = mock_smtp
91 | mock_smtp.__aexit__.return_value = None
92 | mock_smtp.login = AsyncMock()
93 | mock_smtp.send_message = AsyncMock()
94 | return mock_smtp
95 |
96 |
97 | @pytest.fixture
98 | def sample_email_data():
99 | """Fixture for sample email data."""
100 | now = datetime.now()
101 | return {
102 | "subject": "Test Subject",
103 | "from": "[email protected]",
104 | "body": "Test Body",
105 | "date": now,
106 | "attachments": ["attachment.pdf"],
107 | }
108 |
```
--------------------------------------------------------------------------------
/tests/test_models.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import datetime
2 |
3 | from mcp_email_server.emails.models import EmailMetadata, EmailMetadataPageResponse
4 |
5 |
6 | class TestEmailMetadata:
7 | def test_init(self):
8 | """Test initialization with valid data."""
9 | email_data = EmailMetadata(
10 | email_id="123",
11 | subject="Test Subject",
12 | sender="[email protected]",
13 | recipients=["[email protected]"],
14 | date=datetime.now(),
15 | attachments=["file1.txt", "file2.pdf"],
16 | )
17 |
18 | assert email_data.subject == "Test Subject"
19 | assert email_data.sender == "[email protected]"
20 | assert email_data.recipients == ["[email protected]"]
21 | assert isinstance(email_data.date, datetime)
22 | assert email_data.attachments == ["file1.txt", "file2.pdf"]
23 |
24 | def test_from_email(self):
25 | """Test from_email class method."""
26 | now = datetime.now()
27 | email_dict = {
28 | "email_id": "123",
29 | "subject": "Test Subject",
30 | "from": "[email protected]",
31 | "to": ["[email protected]"],
32 | "date": now,
33 | "attachments": ["file1.txt", "file2.pdf"],
34 | }
35 |
36 | email_data = EmailMetadata.from_email(email_dict)
37 |
38 | assert email_data.subject == "Test Subject"
39 | assert email_data.sender == "[email protected]"
40 | assert email_data.recipients == ["[email protected]"]
41 | assert email_data.date == now
42 | assert email_data.attachments == ["file1.txt", "file2.pdf"]
43 |
44 |
45 | class TestEmailMetadataPageResponse:
46 | def test_init(self):
47 | """Test initialization with valid data."""
48 | now = datetime.now()
49 | email_data = EmailMetadata(
50 | email_id="123",
51 | subject="Test Subject",
52 | sender="[email protected]",
53 | recipients=["[email protected]"],
54 | date=now,
55 | attachments=[],
56 | )
57 |
58 | response = EmailMetadataPageResponse(
59 | page=1,
60 | page_size=10,
61 | before=now,
62 | since=None,
63 | subject="Test",
64 | emails=[email_data],
65 | total=1,
66 | )
67 |
68 | assert response.page == 1
69 | assert response.page_size == 10
70 | assert response.before == now
71 | assert response.since is None
72 | assert response.subject == "Test"
73 | assert len(response.emails) == 1
74 | assert response.emails[0] == email_data
75 | assert response.total == 1
76 |
77 | def test_empty_emails(self):
78 | """Test with empty email list."""
79 | response = EmailMetadataPageResponse(
80 | page=1,
81 | page_size=10,
82 | before=None,
83 | since=None,
84 | subject=None,
85 | emails=[],
86 | total=0,
87 | )
88 |
89 | assert response.page == 1
90 | assert response.page_size == 10
91 | assert len(response.emails) == 0
92 | assert response.total == 0
93 |
```
--------------------------------------------------------------------------------
/.github/workflows/on-release-main.yml:
--------------------------------------------------------------------------------
```yaml
1 | name: release-main
2 |
3 | permissions:
4 | contents: write
5 | packages: write
6 |
7 | on:
8 | release:
9 | types: [published]
10 |
11 | jobs:
12 | set-version:
13 | runs-on: ubuntu-24.04
14 | steps:
15 | - uses: actions/checkout@v4
16 |
17 | - name: Export tag
18 | id: vars
19 | run: echo tag=${GITHUB_REF#refs/*/} >> $GITHUB_OUTPUT
20 | if: ${{ github.event_name == 'release' }}
21 |
22 | - name: Update project version
23 | run: |
24 | sed -i "s/^version = \".*\"/version = \"$RELEASE_VERSION\"/" pyproject.toml
25 | env:
26 | RELEASE_VERSION: ${{ steps.vars.outputs.tag }}
27 | if: ${{ github.event_name == 'release' }}
28 |
29 | - name: Upload updated pyproject.toml
30 | uses: actions/upload-artifact@v4
31 | with:
32 | name: pyproject-toml
33 | path: pyproject.toml
34 |
35 | publish:
36 | runs-on: ubuntu-latest
37 | needs: [set-version]
38 | steps:
39 | - name: Check out
40 | uses: actions/checkout@v4
41 |
42 | - name: Set up the environment
43 | uses: ./.github/actions/setup-python-env
44 |
45 | - name: Download updated pyproject.toml
46 | uses: actions/download-artifact@v4
47 | with:
48 | name: pyproject-toml
49 |
50 | - name: Build package
51 | run: uv build
52 |
53 | - name: Publish package
54 | run: uv publish
55 | env:
56 | UV_PUBLISH_TOKEN: ${{ secrets.PYPI_TOKEN }}
57 |
58 | - name: Upload dists to release
59 | uses: svenstaro/upload-release-action@v2
60 | with:
61 | repo_token: ${{ secrets.GITHUB_TOKEN }}
62 | file: dist/*
63 | file_glob: true
64 | tag: ${{ github.ref }}
65 | overwrite: true
66 |
67 | push-image:
68 | runs-on: ubuntu-latest
69 | needs: [set-version]
70 | steps:
71 | - name: Check out
72 | uses: actions/checkout@v4
73 | - name: Export tag
74 | id: vars
75 | run: echo tag=${GITHUB_REF#refs/*/} >> $GITHUB_OUTPUT
76 | if: ${{ github.event_name == 'release' }}
77 | - name: Set up QEMU
78 | uses: docker/setup-qemu-action@v3
79 | - name: Set up Docker Buildx
80 | uses: docker/setup-buildx-action@v3
81 | - name: Login to Github Container Registry
82 | uses: docker/login-action@v3
83 | with:
84 | registry: ghcr.io
85 | username: ai-zerolab
86 | password: ${{ secrets.GITHUB_TOKEN }}
87 | - name: Build and push image
88 | id: docker_build_publish
89 | uses: docker/build-push-action@v5
90 | with:
91 | context: .
92 | platforms: linux/amd64,linux/arm64/v8
93 | cache-from: type=gha
94 | cache-to: type=gha,mode=max
95 | file: ./Dockerfile
96 | push: true
97 | tags: |
98 | ghcr.io/ai-zerolab/mcp-email-server:${{ steps.vars.outputs.tag }}
99 | ghcr.io/ai-zerolab/mcp-email-server:latest
100 |
101 | deploy-docs:
102 | needs: publish
103 | runs-on: ubuntu-latest
104 | steps:
105 | - name: Check out
106 | uses: actions/checkout@v4
107 |
108 | - name: Set up the environment
109 | uses: ./.github/actions/setup-python-env
110 |
111 | - name: Deploy documentation
112 | run: uv run mkdocs gh-deploy --force
113 |
```
--------------------------------------------------------------------------------
/tests/test_dispatcher.py:
--------------------------------------------------------------------------------
```python
1 | from unittest.mock import MagicMock, patch
2 |
3 | import pytest
4 |
5 | from mcp_email_server.config import EmailServer, EmailSettings, ProviderSettings
6 | from mcp_email_server.emails.classic import ClassicEmailHandler
7 | from mcp_email_server.emails.dispatcher import dispatch_handler
8 |
9 |
10 | class TestDispatcher:
11 | def test_dispatch_handler_with_email_settings(self):
12 | """Test dispatch_handler with valid email account."""
13 | # Create test email settings
14 | email_settings = EmailSettings(
15 | account_name="test_account",
16 | full_name="Test User",
17 | email_address="[email protected]",
18 | incoming=EmailServer(
19 | user_name="test_user",
20 | password="test_password",
21 | host="imap.example.com",
22 | port=993,
23 | use_ssl=True,
24 | ),
25 | outgoing=EmailServer(
26 | user_name="test_user",
27 | password="test_password",
28 | host="smtp.example.com",
29 | port=465,
30 | use_ssl=True,
31 | ),
32 | )
33 |
34 | # Mock the get_settings function to return our settings
35 | mock_settings = MagicMock()
36 | mock_settings.get_account.return_value = email_settings
37 |
38 | with patch("mcp_email_server.emails.dispatcher.get_settings", return_value=mock_settings):
39 | # Call the function
40 | handler = dispatch_handler("test_account")
41 |
42 | # Verify the result
43 | assert isinstance(handler, ClassicEmailHandler)
44 | assert handler.email_settings == email_settings
45 |
46 | # Verify get_account was called correctly
47 | mock_settings.get_account.assert_called_once_with("test_account")
48 |
49 | def test_dispatch_handler_with_provider_settings(self):
50 | """Test dispatch_handler with provider account (should raise NotImplementedError)."""
51 | # Create test provider settings
52 | provider_settings = ProviderSettings(
53 | account_name="test_provider",
54 | provider_name="test",
55 | api_key="test_key",
56 | )
57 |
58 | # Mock the get_settings function to return our settings
59 | mock_settings = MagicMock()
60 | mock_settings.get_account.return_value = provider_settings
61 |
62 | with patch("mcp_email_server.emails.dispatcher.get_settings", return_value=mock_settings):
63 | # Call the function and expect NotImplementedError
64 | with pytest.raises(NotImplementedError):
65 | dispatch_handler("test_provider")
66 |
67 | # Verify get_account was called correctly
68 | mock_settings.get_account.assert_called_once_with("test_provider")
69 |
70 | def test_dispatch_handler_with_nonexistent_account(self):
71 | """Test dispatch_handler with non-existent account (should raise ValueError)."""
72 | # Mock the get_settings function to return None for get_account
73 | mock_settings = MagicMock()
74 | mock_settings.get_account.return_value = None
75 | mock_settings.get_accounts.return_value = ["account1", "account2"]
76 |
77 | with patch("mcp_email_server.emails.dispatcher.get_settings", return_value=mock_settings):
78 | # Call the function and expect ValueError
79 | with pytest.raises(ValueError) as excinfo:
80 | dispatch_handler("nonexistent_account")
81 |
82 | # Verify the error message
83 | assert "Account nonexistent_account not found" in str(excinfo.value)
84 |
85 | # Verify get_account was called correctly
86 | mock_settings.get_account.assert_called_once_with("nonexistent_account")
87 | mock_settings.get_accounts.assert_called_once()
88 |
```
--------------------------------------------------------------------------------
/mcp_email_server/app.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import datetime
2 | from typing import Annotated, Literal
3 |
4 | from mcp.server.fastmcp import FastMCP
5 | from pydantic import Field
6 |
7 | from mcp_email_server.config import (
8 | AccountAttributes,
9 | EmailSettings,
10 | ProviderSettings,
11 | get_settings,
12 | )
13 | from mcp_email_server.emails.dispatcher import dispatch_handler
14 | from mcp_email_server.emails.models import EmailContentBatchResponse, EmailMetadataPageResponse
15 |
16 | mcp = FastMCP("email")
17 |
18 |
19 | @mcp.resource("email://{account_name}")
20 | async def get_account(account_name: str) -> EmailSettings | ProviderSettings | None:
21 | settings = get_settings()
22 | return settings.get_account(account_name, masked=True)
23 |
24 |
25 | @mcp.tool(description="List all configured email accounts with masked credentials.")
26 | async def list_available_accounts() -> list[AccountAttributes]:
27 | settings = get_settings()
28 | return [account.masked() for account in settings.get_accounts()]
29 |
30 |
31 | @mcp.tool(description="Add a new email account configuration to the settings.")
32 | async def add_email_account(email: EmailSettings) -> str:
33 | settings = get_settings()
34 | settings.add_email(email)
35 | settings.store()
36 | return f"Successfully added email account '{email.account_name}'"
37 |
38 |
39 | @mcp.tool(
40 | description="List email metadata (email_id, subject, sender, recipients, date) without body content. Returns email_id for use with get_emails_content."
41 | )
42 | async def list_emails_metadata(
43 | account_name: Annotated[str, Field(description="The name of the email account.")],
44 | page: Annotated[
45 | int,
46 | Field(default=1, description="The page number to retrieve (starting from 1)."),
47 | ] = 1,
48 | page_size: Annotated[int, Field(default=10, description="The number of emails to retrieve per page.")] = 10,
49 | before: Annotated[
50 | datetime | None,
51 | Field(default=None, description="Retrieve emails before this datetime (UTC)."),
52 | ] = None,
53 | since: Annotated[
54 | datetime | None,
55 | Field(default=None, description="Retrieve emails since this datetime (UTC)."),
56 | ] = None,
57 | subject: Annotated[str | None, Field(default=None, description="Filter emails by subject.")] = None,
58 | from_address: Annotated[str | None, Field(default=None, description="Filter emails by sender address.")] = None,
59 | to_address: Annotated[
60 | str | None,
61 | Field(default=None, description="Filter emails by recipient address."),
62 | ] = None,
63 | order: Annotated[
64 | Literal["asc", "desc"],
65 | Field(default=None, description="Order emails by field. `asc` or `desc`."),
66 | ] = "desc",
67 | ) -> EmailMetadataPageResponse:
68 | handler = dispatch_handler(account_name)
69 |
70 | return await handler.get_emails_metadata(
71 | page=page,
72 | page_size=page_size,
73 | before=before,
74 | since=since,
75 | subject=subject,
76 | from_address=from_address,
77 | to_address=to_address,
78 | order=order,
79 | )
80 |
81 |
82 | @mcp.tool(
83 | description="Get the full content (including body) of one or more emails by their email_id. Use list_emails_metadata first to get the email_id."
84 | )
85 | async def get_emails_content(
86 | account_name: Annotated[str, Field(description="The name of the email account.")],
87 | email_ids: Annotated[
88 | list[str],
89 | Field(
90 | description="List of email_id to retrieve (obtained from list_emails_metadata). Can be a single email_id or multiple email_ids."
91 | ),
92 | ],
93 | ) -> EmailContentBatchResponse:
94 | handler = dispatch_handler(account_name)
95 | return await handler.get_emails_content(email_ids)
96 |
97 |
98 | @mcp.tool(
99 | description="Send an email using the specified account. Recipient should be a list of email addresses.",
100 | )
101 | async def send_email(
102 | account_name: Annotated[str, Field(description="The name of the email account to send from.")],
103 | recipients: Annotated[list[str], Field(description="A list of recipient email addresses.")],
104 | subject: Annotated[str, Field(description="The subject of the email.")],
105 | body: Annotated[str, Field(description="The body of the email.")],
106 | cc: Annotated[
107 | list[str] | None,
108 | Field(default=None, description="A list of CC email addresses."),
109 | ] = None,
110 | bcc: Annotated[
111 | list[str] | None,
112 | Field(default=None, description="A list of BCC email addresses."),
113 | ] = None,
114 | html: Annotated[
115 | bool,
116 | Field(default=False, description="Whether to send the email as HTML (True) or plain text (False)."),
117 | ] = False,
118 | ) -> str:
119 | handler = dispatch_handler(account_name)
120 | await handler.send_email(recipients, subject, body, cc, bcc, html)
121 | recipient_str = ", ".join(recipients)
122 | return f"Email sent successfully to {recipient_str}"
123 |
```
--------------------------------------------------------------------------------
/mcp_email_server/tools/installer.py:
--------------------------------------------------------------------------------
```python
1 | import json
2 | import os
3 | import platform
4 | import shutil
5 | import sys
6 | from pathlib import Path
7 |
8 | from jinja2 import Template
9 |
10 | _HERE = Path(__file__).parent
11 | CLAUDE_DESKTOP_CONFIG_TEMPLATE = _HERE / "claude_desktop_config.json"
12 |
13 | system = platform.system()
14 | if system == "Darwin":
15 | CLAUDE_DESKTOP_CONFIG_PATH = os.path.expanduser("~/Library/Application Support/Claude/claude_desktop_config.json")
16 | elif system == "Windows":
17 | CLAUDE_DESKTOP_CONFIG_PATH = os.path.join(os.environ["APPDATA"], "Claude", "claude_desktop_config.json")
18 | else:
19 | CLAUDE_DESKTOP_CONFIG_PATH = None
20 |
21 |
22 | def get_endpoint_path() -> str:
23 | """
24 | Find the path to the mcp-email-server script.
25 | Similar to the 'which' command in Unix-like systems.
26 |
27 | Returns:
28 | str: The full path to the mcp-email-server script
29 | """
30 | # First try using shutil.which to find the script in PATH
31 | script_path = shutil.which("mcp-email-server")
32 | if script_path:
33 | return script_path
34 |
35 | # If not found in PATH, try to find it in the current Python environment
36 | # This handles cases where the script is installed but not in PATH
37 | bin_dir = Path(sys.executable).parent
38 | possible_paths = [
39 | bin_dir / "mcp-email-server",
40 | bin_dir / "mcp-email-server.exe", # For Windows
41 | ]
42 |
43 | for path in possible_paths:
44 | if path.exists():
45 | return str(path)
46 |
47 | # If we can't find it, return the script name and hope it's in PATH when executed
48 | return "mcp-email-server"
49 |
50 |
51 | def install_claude_desktop():
52 | # Read the template config
53 | template_content = CLAUDE_DESKTOP_CONFIG_TEMPLATE.read_text()
54 | rendered_content = Template(template_content).render(ENTRYPOINT=get_endpoint_path())
55 | template_config = json.loads(rendered_content)
56 | if not CLAUDE_DESKTOP_CONFIG_PATH:
57 | raise NotImplementedError
58 |
59 | # Read the existing config file or create an empty JSON object
60 | try:
61 | with open(CLAUDE_DESKTOP_CONFIG_PATH) as f:
62 | existing_config = json.load(f)
63 | except FileNotFoundError:
64 | existing_config = {}
65 |
66 | # Merge the template config into the existing config
67 | if "mcpServers" not in existing_config:
68 | existing_config["mcpServers"] = {}
69 | existing_config["mcpServers"].update(template_config["mcpServers"])
70 |
71 | # Write the merged config back to the Claude config file
72 | os.makedirs(os.path.dirname(CLAUDE_DESKTOP_CONFIG_PATH), exist_ok=True)
73 | with open(CLAUDE_DESKTOP_CONFIG_PATH, "w") as f:
74 | json.dump(existing_config, f, indent=4)
75 |
76 |
77 | def uninstall_claude_desktop():
78 | if not CLAUDE_DESKTOP_CONFIG_PATH:
79 | raise NotImplementedError
80 | try:
81 | with open(CLAUDE_DESKTOP_CONFIG_PATH) as f:
82 | existing_config = json.load(f)
83 | except FileNotFoundError:
84 | return
85 |
86 | if "mcpServers" not in existing_config:
87 | return
88 |
89 | if "zerolib-email" in existing_config["mcpServers"]:
90 | del existing_config["mcpServers"]["zerolib-email"]
91 |
92 | with open(CLAUDE_DESKTOP_CONFIG_PATH, "w") as f:
93 | json.dump(existing_config, f, indent=4)
94 |
95 |
96 | def is_installed() -> bool:
97 | """
98 | Check if the MCP email server is installed in the Claude desktop configuration.
99 |
100 | Returns:
101 | bool: True if installed, False otherwise
102 | """
103 | if not CLAUDE_DESKTOP_CONFIG_PATH:
104 | return False
105 |
106 | try:
107 | with open(CLAUDE_DESKTOP_CONFIG_PATH) as f:
108 | config = json.load(f)
109 |
110 | return "mcpServers" in config and "zerolib-email" in config["mcpServers"]
111 | except (FileNotFoundError, json.JSONDecodeError):
112 | return False
113 |
114 |
115 | def need_update() -> bool:
116 | """
117 | Check if the installed configuration needs to be updated.
118 |
119 | Returns:
120 | bool: True if an update is needed, False otherwise
121 | """
122 | if not is_installed():
123 | return True
124 |
125 | try:
126 | # Get the template config
127 | template_content = CLAUDE_DESKTOP_CONFIG_TEMPLATE.read_text()
128 | rendered_content = Template(template_content).render(ENTRYPOINT=get_endpoint_path())
129 | template_config = json.loads(rendered_content)
130 |
131 | # Get the installed config
132 | with open(CLAUDE_DESKTOP_CONFIG_PATH) as f:
133 | installed_config = json.load(f)
134 |
135 | # Compare the relevant parts of the configs
136 | template_server = template_config["mcpServers"]["zerolib-email"]
137 | installed_server = installed_config["mcpServers"]["zerolib-email"]
138 |
139 | # Check if any key configuration elements differ
140 | return (
141 | template_server.get("command") != installed_server.get("command")
142 | or template_server.get("args") != installed_server.get("args")
143 | or template_server.get("env") != installed_server.get("env")
144 | )
145 | except (FileNotFoundError, json.JSONDecodeError, KeyError):
146 | # If any error occurs during comparison, suggest an update
147 | return True
148 |
149 |
150 | def get_claude_desktop_config() -> str:
151 | if not CLAUDE_DESKTOP_CONFIG_PATH:
152 | raise NotImplementedError
153 |
154 | with open(CLAUDE_DESKTOP_CONFIG_PATH) as f:
155 | return f.read()
156 |
```
--------------------------------------------------------------------------------
/tests/test_classic_handler.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import datetime
2 | from unittest.mock import AsyncMock, patch
3 |
4 | import pytest
5 |
6 | from mcp_email_server.config import EmailServer, EmailSettings
7 | from mcp_email_server.emails.classic import ClassicEmailHandler, EmailClient
8 | from mcp_email_server.emails.models import EmailMetadata, EmailMetadataPageResponse
9 |
10 |
11 | @pytest.fixture
12 | def email_settings():
13 | return EmailSettings(
14 | account_name="test_account",
15 | full_name="Test User",
16 | email_address="[email protected]",
17 | incoming=EmailServer(
18 | user_name="test_user",
19 | password="test_password",
20 | host="imap.example.com",
21 | port=993,
22 | use_ssl=True,
23 | ),
24 | outgoing=EmailServer(
25 | user_name="test_user",
26 | password="test_password",
27 | host="smtp.example.com",
28 | port=465,
29 | use_ssl=True,
30 | ),
31 | )
32 |
33 |
34 | @pytest.fixture
35 | def classic_handler(email_settings):
36 | return ClassicEmailHandler(email_settings)
37 |
38 |
39 | class TestClassicEmailHandler:
40 | def test_init(self, email_settings):
41 | """Test initialization of ClassicEmailHandler."""
42 | handler = ClassicEmailHandler(email_settings)
43 |
44 | assert handler.email_settings == email_settings
45 | assert isinstance(handler.incoming_client, EmailClient)
46 | assert isinstance(handler.outgoing_client, EmailClient)
47 |
48 | # Check that clients are initialized correctly
49 | assert handler.incoming_client.email_server == email_settings.incoming
50 | assert handler.outgoing_client.email_server == email_settings.outgoing
51 | assert handler.outgoing_client.sender == f"{email_settings.full_name} <{email_settings.email_address}>"
52 |
53 | @pytest.mark.asyncio
54 | async def test_get_emails(self, classic_handler):
55 | """Test get_emails method."""
56 | # Create test data
57 | now = datetime.now()
58 | email_data = {
59 | "email_id": "123",
60 | "subject": "Test Subject",
61 | "from": "[email protected]",
62 | "to": ["[email protected]"],
63 | "date": now,
64 | "attachments": [],
65 | }
66 |
67 | # Mock the get_emails_stream method to yield our test data
68 | mock_stream = AsyncMock()
69 | mock_stream.__aiter__.return_value = [email_data]
70 |
71 | # Mock the get_email_count method
72 | mock_count = AsyncMock(return_value=1)
73 |
74 | # Apply the mocks
75 | with patch.object(classic_handler.incoming_client, "get_emails_metadata_stream", return_value=mock_stream):
76 | with patch.object(classic_handler.incoming_client, "get_email_count", mock_count):
77 | # Call the method
78 | result = await classic_handler.get_emails_metadata(
79 | page=1,
80 | page_size=10,
81 | before=now,
82 | since=None,
83 | subject="Test",
84 | from_address="[email protected]",
85 | to_address=None,
86 | )
87 |
88 | # Verify the result
89 | assert isinstance(result, EmailMetadataPageResponse)
90 | assert result.page == 1
91 | assert result.page_size == 10
92 | assert result.before == now
93 | assert result.since is None
94 | assert result.subject == "Test"
95 | assert len(result.emails) == 1
96 | assert isinstance(result.emails[0], EmailMetadata)
97 | assert result.emails[0].subject == "Test Subject"
98 | assert result.emails[0].sender == "[email protected]"
99 | assert result.emails[0].date == now
100 | assert result.emails[0].attachments == []
101 | assert result.total == 1
102 |
103 | # Verify the client methods were called correctly
104 | classic_handler.incoming_client.get_emails_metadata_stream.assert_called_once_with(
105 | 1, 10, now, None, "Test", "[email protected]", None, "desc"
106 | )
107 | mock_count.assert_called_once_with(
108 | now, None, "Test", from_address="[email protected]", to_address=None
109 | )
110 |
111 | @pytest.mark.asyncio
112 | async def test_send_email(self, classic_handler):
113 | """Test send_email method."""
114 | # Mock the outgoing_client.send_email method
115 | mock_send = AsyncMock()
116 |
117 | # Apply the mock
118 | with patch.object(classic_handler.outgoing_client, "send_email", mock_send):
119 | # Call the method
120 | await classic_handler.send_email(
121 | recipients=["[email protected]"],
122 | subject="Test Subject",
123 | body="Test Body",
124 | cc=["[email protected]"],
125 | bcc=["[email protected]"],
126 | )
127 |
128 | # Verify the client method was called correctly
129 | mock_send.assert_called_once_with(
130 | ["[email protected]"],
131 | "Test Subject",
132 | "Test Body",
133 | ["[email protected]"],
134 | ["[email protected]"],
135 | False,
136 | )
137 |
```
--------------------------------------------------------------------------------
/tests/test_mcp_tools.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import datetime
2 | from unittest.mock import AsyncMock, MagicMock, patch
3 |
4 | import pytest
5 |
6 | from mcp_email_server.app import (
7 | add_email_account,
8 | get_emails_content,
9 | list_available_accounts,
10 | list_emails_metadata,
11 | send_email,
12 | )
13 | from mcp_email_server.config import EmailServer, EmailSettings, ProviderSettings
14 | from mcp_email_server.emails.models import (
15 | EmailBodyResponse,
16 | EmailContentBatchResponse,
17 | EmailMetadata,
18 | EmailMetadataPageResponse,
19 | )
20 |
21 |
22 | class TestMcpTools:
23 | @pytest.mark.asyncio
24 | async def test_list_available_accounts(self):
25 | """Test list_available_accounts MCP tool."""
26 | # Create test accounts
27 | email_settings = EmailSettings(
28 | account_name="test_email",
29 | full_name="Test User",
30 | email_address="[email protected]",
31 | incoming=EmailServer(
32 | user_name="test_user",
33 | password="test_password",
34 | host="imap.example.com",
35 | port=993,
36 | use_ssl=True,
37 | ),
38 | outgoing=EmailServer(
39 | user_name="test_user",
40 | password="test_password",
41 | host="smtp.example.com",
42 | port=465,
43 | use_ssl=True,
44 | ),
45 | )
46 |
47 | provider_settings = ProviderSettings(
48 | account_name="test_provider",
49 | provider_name="test",
50 | api_key="test_key",
51 | )
52 |
53 | # Mock the get_settings function
54 | mock_settings = MagicMock()
55 | mock_settings.get_accounts.return_value = [email_settings, provider_settings]
56 |
57 | with patch("mcp_email_server.app.get_settings", return_value=mock_settings):
58 | # Call the function
59 | result = await list_available_accounts()
60 |
61 | # Verify the result
62 | assert len(result) == 2
63 | assert result[0].account_name == "test_email"
64 | assert result[1].account_name == "test_provider"
65 |
66 | # Verify get_accounts was called correctly
67 | mock_settings.get_accounts.assert_called_once()
68 |
69 | @pytest.mark.asyncio
70 | async def test_add_email_account(self):
71 | """Test add_email_account MCP tool."""
72 | # Create test email settings
73 | email_settings = EmailSettings(
74 | account_name="test_account",
75 | full_name="Test User",
76 | email_address="[email protected]",
77 | incoming=EmailServer(
78 | user_name="test_user",
79 | password="test_password",
80 | host="imap.example.com",
81 | port=993,
82 | use_ssl=True,
83 | ),
84 | outgoing=EmailServer(
85 | user_name="test_user",
86 | password="test_password",
87 | host="smtp.example.com",
88 | port=465,
89 | use_ssl=True,
90 | ),
91 | )
92 |
93 | # Mock the get_settings function
94 | mock_settings = MagicMock()
95 |
96 | with patch("mcp_email_server.app.get_settings", return_value=mock_settings):
97 | # Call the function
98 | result = await add_email_account(email_settings)
99 |
100 | # Verify the return value
101 | assert result == "Successfully added email account 'test_account'"
102 |
103 | # Verify add_email and store were called correctly
104 | mock_settings.add_email.assert_called_once_with(email_settings)
105 | mock_settings.store.assert_called_once()
106 |
107 | @pytest.mark.asyncio
108 | async def test_list_emails_metadata(self):
109 | """Test list_emails_metadata MCP tool."""
110 | # Create test data
111 | now = datetime.now()
112 | email_metadata = EmailMetadata(
113 | email_id="12345",
114 | subject="Test Subject",
115 | sender="[email protected]",
116 | recipients=["[email protected]"],
117 | date=now,
118 | attachments=[],
119 | )
120 |
121 | email_metadata_page = EmailMetadataPageResponse(
122 | page=1,
123 | page_size=10,
124 | before=now,
125 | since=None,
126 | subject="Test",
127 | emails=[email_metadata],
128 | total=1,
129 | )
130 |
131 | # Mock the dispatch_handler function
132 | mock_handler = AsyncMock()
133 | mock_handler.get_emails_metadata.return_value = email_metadata_page
134 |
135 | with patch("mcp_email_server.app.dispatch_handler", return_value=mock_handler):
136 | # Call the function
137 | result = await list_emails_metadata(
138 | account_name="test_account",
139 | page=1,
140 | page_size=10,
141 | before=now,
142 | since=None,
143 | subject="Test",
144 | from_address="[email protected]",
145 | to_address=None,
146 | )
147 |
148 | # Verify the result
149 | assert result == email_metadata_page
150 | assert result.page == 1
151 | assert result.page_size == 10
152 | assert result.before == now
153 | assert result.subject == "Test"
154 | assert len(result.emails) == 1
155 | assert result.emails[0].subject == "Test Subject"
156 | assert result.emails[0].email_id == "12345"
157 |
158 | # Verify dispatch_handler and get_emails_metadata were called correctly
159 | mock_handler.get_emails_metadata.assert_called_once_with(
160 | page=1,
161 | page_size=10,
162 | before=now,
163 | since=None,
164 | subject="Test",
165 | from_address="[email protected]",
166 | to_address=None,
167 | order="desc",
168 | )
169 |
170 | @pytest.mark.asyncio
171 | async def test_get_emails_content_single(self):
172 | """Test get_emails_content MCP tool with single email."""
173 | # Create test data
174 | now = datetime.now()
175 | email_body = EmailBodyResponse(
176 | email_id="12345",
177 | subject="Test Subject",
178 | sender="[email protected]",
179 | recipients=["[email protected]"],
180 | date=now,
181 | body="This is the test email body content.",
182 | attachments=["attachment1.pdf"],
183 | )
184 |
185 | batch_response = EmailContentBatchResponse(
186 | emails=[email_body],
187 | requested_count=1,
188 | retrieved_count=1,
189 | failed_ids=[],
190 | )
191 |
192 | # Mock the dispatch_handler function
193 | mock_handler = AsyncMock()
194 | mock_handler.get_emails_content.return_value = batch_response
195 |
196 | with patch("mcp_email_server.app.dispatch_handler", return_value=mock_handler):
197 | # Call the function
198 | result = await get_emails_content(
199 | account_name="test_account",
200 | email_ids=["12345"],
201 | )
202 |
203 | # Verify the result
204 | assert result == batch_response
205 | assert result.requested_count == 1
206 | assert result.retrieved_count == 1
207 | assert len(result.failed_ids) == 0
208 | assert len(result.emails) == 1
209 | assert result.emails[0].email_id == "12345"
210 | assert result.emails[0].subject == "Test Subject"
211 |
212 | # Verify dispatch_handler and get_emails_content were called correctly
213 | mock_handler.get_emails_content.assert_called_once_with(["12345"])
214 |
215 | @pytest.mark.asyncio
216 | async def test_get_emails_content_batch(self):
217 | """Test get_emails_content MCP tool with multiple emails."""
218 | # Create test data
219 | now = datetime.now()
220 | email1 = EmailBodyResponse(
221 | email_id="12345",
222 | subject="Test Subject 1",
223 | sender="[email protected]",
224 | recipients=["[email protected]"],
225 | date=now,
226 | body="This is the first test email body content.",
227 | attachments=[],
228 | )
229 |
230 | email2 = EmailBodyResponse(
231 | email_id="12346",
232 | subject="Test Subject 2",
233 | sender="[email protected]",
234 | recipients=["[email protected]"],
235 | date=now,
236 | body="This is the second test email body content.",
237 | attachments=["attachment1.pdf"],
238 | )
239 |
240 | batch_response = EmailContentBatchResponse(
241 | emails=[email1, email2],
242 | requested_count=3,
243 | retrieved_count=2,
244 | failed_ids=["12347"],
245 | )
246 |
247 | # Mock the dispatch_handler function
248 | mock_handler = AsyncMock()
249 | mock_handler.get_emails_content.return_value = batch_response
250 |
251 | with patch("mcp_email_server.app.dispatch_handler", return_value=mock_handler):
252 | # Call the function
253 | result = await get_emails_content(
254 | account_name="test_account",
255 | email_ids=["12345", "12346", "12347"],
256 | )
257 |
258 | # Verify the result
259 | assert result == batch_response
260 | assert result.requested_count == 3
261 | assert result.retrieved_count == 2
262 | assert len(result.failed_ids) == 1
263 | assert result.failed_ids[0] == "12347"
264 | assert len(result.emails) == 2
265 | assert result.emails[0].email_id == "12345"
266 | assert result.emails[1].email_id == "12346"
267 |
268 | # Verify dispatch_handler and get_emails_content were called correctly
269 | mock_handler.get_emails_content.assert_called_once_with(["12345", "12346", "12347"])
270 |
271 | @pytest.mark.asyncio
272 | async def test_send_email(self):
273 | """Test send_email MCP tool."""
274 | # Mock the dispatch_handler function
275 | mock_handler = AsyncMock()
276 |
277 | with patch("mcp_email_server.app.dispatch_handler", return_value=mock_handler):
278 | # Call the function
279 | result = await send_email(
280 | account_name="test_account",
281 | recipients=["[email protected]"],
282 | subject="Test Subject",
283 | body="Test Body",
284 | cc=["[email protected]"],
285 | bcc=["[email protected]"],
286 | )
287 |
288 | # Verify the return value
289 | assert result == "Email sent successfully to [email protected]"
290 |
291 | # Verify send_email was called correctly
292 | mock_handler.send_email.assert_called_once_with(
293 | ["[email protected]"],
294 | "Test Subject",
295 | "Test Body",
296 | ["[email protected]"],
297 | ["[email protected]"],
298 | False,
299 | )
300 |
```
--------------------------------------------------------------------------------
/tests/test_email_client.py:
--------------------------------------------------------------------------------
```python
1 | import asyncio
2 | import email
3 | from datetime import datetime
4 | from email.mime.text import MIMEText
5 | from unittest.mock import AsyncMock, MagicMock, patch
6 |
7 | import pytest
8 |
9 | from mcp_email_server.config import EmailServer
10 | from mcp_email_server.emails.classic import EmailClient
11 |
12 |
13 | @pytest.fixture
14 | def email_server():
15 | return EmailServer(
16 | user_name="test_user",
17 | password="test_password",
18 | host="imap.example.com",
19 | port=993,
20 | use_ssl=True,
21 | )
22 |
23 |
24 | @pytest.fixture
25 | def email_client(email_server):
26 | return EmailClient(email_server, sender="Test User <[email protected]>")
27 |
28 |
29 | class TestEmailClient:
30 | def test_init(self, email_server):
31 | """Test initialization of EmailClient."""
32 | client = EmailClient(email_server)
33 | assert client.email_server == email_server
34 | assert client.sender == email_server.user_name
35 | assert client.smtp_use_tls is True
36 | assert client.smtp_start_tls is False
37 |
38 | # Test with custom sender
39 | custom_sender = "Custom <[email protected]>"
40 | client = EmailClient(email_server, sender=custom_sender)
41 | assert client.sender == custom_sender
42 |
43 | def test_parse_email_data_plain(self):
44 | """Test parsing plain text email."""
45 | # Create a simple plain text email
46 | msg = MIMEText("This is a test email body")
47 | msg["Subject"] = "Test Subject"
48 | msg["From"] = "[email protected]"
49 | msg["To"] = "[email protected]"
50 | msg["Date"] = email.utils.formatdate()
51 |
52 | raw_email = msg.as_bytes()
53 |
54 | client = EmailClient(MagicMock())
55 | result = client._parse_email_data(raw_email)
56 |
57 | assert result["subject"] == "Test Subject"
58 | assert result["from"] == "[email protected]"
59 | assert result["body"] == "This is a test email body"
60 | assert isinstance(result["date"], datetime)
61 | assert result["attachments"] == []
62 |
63 | def test_parse_email_data_with_attachments(self):
64 | """Test parsing email with attachments."""
65 | # This would require creating a multipart email with attachments
66 | # For simplicity, we'll mock the email parsing
67 | with patch("email.parser.BytesParser.parsebytes") as mock_parse:
68 | mock_email = MagicMock()
69 | mock_email.get.side_effect = lambda x, default=None: {
70 | "Subject": "Test Subject",
71 | "From": "[email protected]",
72 | "Date": email.utils.formatdate(),
73 | }.get(x, default)
74 | mock_email.is_multipart.return_value = True
75 |
76 | # Mock parts
77 | text_part = MagicMock()
78 | text_part.get_content_type.return_value = "text/plain"
79 | text_part.get.return_value = "" # Not an attachment
80 | text_part.get_payload.return_value = b"This is the email body"
81 | text_part.get_content_charset.return_value = "utf-8"
82 |
83 | attachment_part = MagicMock()
84 | attachment_part.get_content_type.return_value = "application/pdf"
85 | attachment_part.get.return_value = "attachment; filename=test.pdf"
86 | attachment_part.get_filename.return_value = "test.pdf"
87 |
88 | mock_email.walk.return_value = [text_part, attachment_part]
89 | mock_parse.return_value = mock_email
90 |
91 | client = EmailClient(MagicMock())
92 | result = client._parse_email_data(b"dummy email content")
93 |
94 | assert result["subject"] == "Test Subject"
95 | assert result["from"] == "[email protected]"
96 | assert result["body"] == "This is the email body"
97 | assert isinstance(result["date"], datetime)
98 | assert result["attachments"] == ["test.pdf"]
99 |
100 | def test_build_search_criteria(self):
101 | """Test building search criteria for IMAP."""
102 | # Test with no criteria (should return ["ALL"])
103 | criteria = EmailClient._build_search_criteria()
104 | assert criteria == ["ALL"]
105 |
106 | # Test with before date
107 | before_date = datetime(2023, 1, 1)
108 | criteria = EmailClient._build_search_criteria(before=before_date)
109 | assert criteria == ["BEFORE", "01-JAN-2023"]
110 |
111 | # Test with since date
112 | since_date = datetime(2023, 1, 1)
113 | criteria = EmailClient._build_search_criteria(since=since_date)
114 | assert criteria == ["SINCE", "01-JAN-2023"]
115 |
116 | # Test with subject
117 | criteria = EmailClient._build_search_criteria(subject="Test")
118 | assert criteria == ["SUBJECT", "Test"]
119 |
120 | # Test with body
121 | criteria = EmailClient._build_search_criteria(body="Test")
122 | assert criteria == ["BODY", "Test"]
123 |
124 | # Test with text
125 | criteria = EmailClient._build_search_criteria(text="Test")
126 | assert criteria == ["TEXT", "Test"]
127 |
128 | # Test with from_address
129 | criteria = EmailClient._build_search_criteria(from_address="[email protected]")
130 | assert criteria == ["FROM", "[email protected]"]
131 |
132 | # Test with to_address
133 | criteria = EmailClient._build_search_criteria(to_address="[email protected]")
134 | assert criteria == ["TO", "[email protected]"]
135 |
136 | # Test with multiple criteria
137 | criteria = EmailClient._build_search_criteria(
138 | subject="Test", from_address="[email protected]", since=datetime(2023, 1, 1)
139 | )
140 | assert criteria == ["SINCE", "01-JAN-2023", "SUBJECT", "Test", "FROM", "[email protected]"]
141 |
142 | @pytest.mark.asyncio
143 | async def test_get_emails_stream(self, email_client):
144 | """Test getting emails stream."""
145 | # Mock IMAP client
146 | mock_imap = AsyncMock()
147 | mock_imap._client_task = asyncio.Future()
148 | mock_imap._client_task.set_result(None)
149 | mock_imap.wait_hello_from_server = AsyncMock()
150 | mock_imap.login = AsyncMock()
151 | mock_imap.select = AsyncMock()
152 | mock_imap.search = AsyncMock(return_value=(None, [b"1 2 3"]))
153 | mock_imap.uid_search = AsyncMock(return_value=(None, [b"1 2 3"]))
154 | mock_imap.fetch = AsyncMock(return_value=(None, [b"HEADER", bytearray(b"EMAIL CONTENT")]))
155 | # Create a simple email with headers for testing
156 | test_email = b"""From: [email protected]\r
157 | To: [email protected]\r
158 | Subject: Test Subject\r
159 | Date: Mon, 1 Jan 2024 00:00:00 +0000\r
160 | \r
161 | This is the email body."""
162 | mock_imap.uid = AsyncMock(
163 | return_value=(None, [b"1 FETCH (UID 1 RFC822 {%d}" % len(test_email), bytearray(test_email)])
164 | )
165 | mock_imap.logout = AsyncMock()
166 |
167 | # Mock IMAP class
168 | with patch.object(email_client, "imap_class", return_value=mock_imap):
169 | # Mock _parse_email_data
170 | with patch.object(email_client, "_parse_email_data") as mock_parse:
171 | mock_parse.return_value = {
172 | "subject": "Test Subject",
173 | "from": "[email protected]",
174 | "body": "Test Body",
175 | "date": datetime.now(),
176 | "attachments": [],
177 | }
178 |
179 | emails = []
180 | async for email_data in email_client.get_emails_metadata_stream(page=1, page_size=10):
181 | emails.append(email_data)
182 |
183 | # We should get 3 emails (from the mocked search result "1 2 3")
184 | assert len(emails) == 3
185 | assert emails[0]["subject"] == "Test Subject"
186 | assert emails[0]["from"] == "[email protected]"
187 |
188 | # Verify IMAP methods were called correctly
189 | mock_imap.login.assert_called_once_with(
190 | email_client.email_server.user_name, email_client.email_server.password
191 | )
192 | mock_imap.select.assert_called_once_with("INBOX")
193 | mock_imap.uid_search.assert_called_once_with("ALL")
194 | assert mock_imap.uid.call_count == 3
195 | mock_imap.logout.assert_called_once()
196 |
197 | @pytest.mark.asyncio
198 | async def test_get_email_count(self, email_client):
199 | """Test getting email count."""
200 | # Mock IMAP client
201 | mock_imap = AsyncMock()
202 | mock_imap._client_task = asyncio.Future()
203 | mock_imap._client_task.set_result(None)
204 | mock_imap.wait_hello_from_server = AsyncMock()
205 | mock_imap.login = AsyncMock()
206 | mock_imap.select = AsyncMock()
207 | mock_imap.search = AsyncMock(return_value=(None, [b"1 2 3 4 5"]))
208 | mock_imap.uid_search = AsyncMock(return_value=(None, [b"1 2 3 4 5"]))
209 | mock_imap.logout = AsyncMock()
210 |
211 | # Mock IMAP class
212 | with patch.object(email_client, "imap_class", return_value=mock_imap):
213 | count = await email_client.get_email_count()
214 |
215 | assert count == 5
216 |
217 | # Verify IMAP methods were called correctly
218 | mock_imap.login.assert_called_once_with(
219 | email_client.email_server.user_name, email_client.email_server.password
220 | )
221 | mock_imap.select.assert_called_once_with("INBOX")
222 | mock_imap.uid_search.assert_called_once_with("ALL")
223 | mock_imap.logout.assert_called_once()
224 |
225 | @pytest.mark.asyncio
226 | async def test_send_email(self, email_client):
227 | """Test sending email."""
228 | # Mock SMTP client
229 | mock_smtp = AsyncMock()
230 | mock_smtp.__aenter__.return_value = mock_smtp
231 | mock_smtp.__aexit__.return_value = None
232 | mock_smtp.login = AsyncMock()
233 | mock_smtp.send_message = AsyncMock()
234 |
235 | with patch("aiosmtplib.SMTP", return_value=mock_smtp):
236 | await email_client.send_email(
237 | recipients=["[email protected]"],
238 | subject="Test Subject",
239 | body="Test Body",
240 | cc=["[email protected]"],
241 | bcc=["[email protected]"],
242 | )
243 |
244 | # Verify SMTP methods were called correctly
245 | mock_smtp.login.assert_called_once_with(
246 | email_client.email_server.user_name, email_client.email_server.password
247 | )
248 | mock_smtp.send_message.assert_called_once()
249 |
250 | # Check that the message was constructed correctly
251 | call_args = mock_smtp.send_message.call_args
252 | msg = call_args[0][0]
253 | recipients = call_args[1]["recipients"]
254 |
255 | assert msg["Subject"] == "Test Subject"
256 | assert msg["From"] == email_client.sender
257 | assert msg["To"] == "[email protected]"
258 | assert msg["Cc"] == "[email protected]"
259 | assert "Bcc" not in msg # BCC should not be in headers
260 |
261 | # Check that all recipients are included in the SMTP call
262 | assert "[email protected]" in recipients
263 | assert "[email protected]" in recipients
264 | assert "[email protected]" in recipients
265 |
```
--------------------------------------------------------------------------------
/tests/test_env_config_coverage.py:
--------------------------------------------------------------------------------
```python
1 | """Focused tests to achieve patch coverage for environment variable configuration."""
2 |
3 | import os
4 |
5 | from mcp_email_server.config import EmailSettings, Settings
6 |
7 |
8 | def test_from_env_missing_email_and_password(monkeypatch):
9 | """Test return None when email/password missing - covers line 138."""
10 | # No environment variables set
11 | result = EmailSettings.from_env()
12 | assert result is None
13 |
14 | # Only email, no password
15 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
16 | result = EmailSettings.from_env()
17 | assert result is None
18 |
19 | # Only password, no email
20 | monkeypatch.delenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", raising=False)
21 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "pass")
22 | result = EmailSettings.from_env()
23 | assert result is None
24 |
25 |
26 | def test_from_env_missing_hosts_warning(monkeypatch):
27 | """Test logger.warning for missing hosts - covers lines 154-156."""
28 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
29 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "pass")
30 |
31 | # Missing both hosts
32 | result = EmailSettings.from_env()
33 | assert result is None
34 |
35 | # Missing SMTP host
36 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.test.com")
37 | result = EmailSettings.from_env()
38 | assert result is None
39 |
40 | # Missing IMAP host
41 | monkeypatch.delenv("MCP_EMAIL_SERVER_IMAP_HOST")
42 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.test.com")
43 | result = EmailSettings.from_env()
44 | assert result is None
45 |
46 |
47 | def test_from_env_exception_handling(monkeypatch):
48 | """Test exception handling in try/except - covers lines 177-179."""
49 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
50 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "pass")
51 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.test.com")
52 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.test.com")
53 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_PORT", "invalid") # Will cause ValueError
54 |
55 | result = EmailSettings.from_env()
56 | assert result is None
57 |
58 |
59 | def test_from_env_success_with_all_defaults(monkeypatch):
60 | """Test successful creation with defaults - covers lines 147-176."""
61 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
62 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "pass")
63 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.example.com")
64 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.example.com")
65 |
66 | result = EmailSettings.from_env()
67 | assert result is not None
68 | assert result.account_name == "default"
69 | assert result.full_name == "user"
70 | assert result.email_address == "[email protected]"
71 | assert result.incoming.user_name == "[email protected]"
72 | assert result.incoming.port == 993
73 | assert result.outgoing.port == 465
74 |
75 |
76 | def test_from_env_with_all_vars_set(monkeypatch):
77 | """Test with all environment variables set - covers parse_bool branches."""
78 | env_vars = {
79 | "MCP_EMAIL_SERVER_ACCOUNT_NAME": "myaccount",
80 | "MCP_EMAIL_SERVER_FULL_NAME": "John Doe",
81 | "MCP_EMAIL_SERVER_EMAIL_ADDRESS": "[email protected]",
82 | "MCP_EMAIL_SERVER_USER_NAME": "johnuser",
83 | "MCP_EMAIL_SERVER_PASSWORD": "pass123",
84 | "MCP_EMAIL_SERVER_IMAP_HOST": "imap.example.com",
85 | "MCP_EMAIL_SERVER_IMAP_PORT": "143",
86 | "MCP_EMAIL_SERVER_IMAP_SSL": "false",
87 | "MCP_EMAIL_SERVER_SMTP_HOST": "smtp.example.com",
88 | "MCP_EMAIL_SERVER_SMTP_PORT": "587",
89 | "MCP_EMAIL_SERVER_SMTP_SSL": "no",
90 | "MCP_EMAIL_SERVER_SMTP_START_SSL": "yes",
91 | "MCP_EMAIL_SERVER_IMAP_USER_NAME": "imap_john",
92 | "MCP_EMAIL_SERVER_IMAP_PASSWORD": "imap_pass",
93 | "MCP_EMAIL_SERVER_SMTP_USER_NAME": "smtp_john",
94 | "MCP_EMAIL_SERVER_SMTP_PASSWORD": "smtp_pass",
95 | }
96 |
97 | for key, value in env_vars.items():
98 | monkeypatch.setenv(key, value)
99 |
100 | result = EmailSettings.from_env()
101 | assert result is not None
102 | assert result.account_name == "myaccount"
103 | assert result.full_name == "John Doe"
104 | assert result.incoming.user_name == "imap_john"
105 | assert result.incoming.password == "imap_pass" # noqa: S105
106 | assert result.incoming.port == 143
107 | assert result.incoming.use_ssl is False
108 | assert result.outgoing.user_name == "smtp_john"
109 | assert result.outgoing.password == "smtp_pass" # noqa: S105
110 | assert result.outgoing.port == 587
111 | assert result.outgoing.use_ssl is False
112 | assert result.outgoing.start_ssl is True
113 |
114 |
115 | def test_from_env_boolean_parsing_variations(monkeypatch):
116 | """Test various boolean value parsing - covers parse_bool function."""
117 | base_env = {
118 | "MCP_EMAIL_SERVER_EMAIL_ADDRESS": "[email protected]",
119 | "MCP_EMAIL_SERVER_PASSWORD": "pass",
120 | "MCP_EMAIL_SERVER_IMAP_HOST": "imap.test.com",
121 | "MCP_EMAIL_SERVER_SMTP_HOST": "smtp.test.com",
122 | }
123 |
124 | # Test "1" = true
125 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_SSL", "1")
126 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_SSL", "0")
127 | for key, value in base_env.items():
128 | monkeypatch.setenv(key, value)
129 |
130 | result = EmailSettings.from_env()
131 | assert result.incoming.use_ssl is True
132 | assert result.outgoing.use_ssl is False
133 |
134 | # Test "on"/"off"
135 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_SSL", "on")
136 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_START_SSL", "off")
137 |
138 | result = EmailSettings.from_env()
139 | assert result.incoming.use_ssl is True
140 | assert result.outgoing.start_ssl is False
141 |
142 |
143 | def test_settings_init_no_env(monkeypatch, tmp_path):
144 | """Test Settings.__init__ when no env vars - covers line 211 false branch."""
145 | config_file = tmp_path / "empty.toml"
146 | config_file.write_text("")
147 | monkeypatch.setenv("MCP_EMAIL_SERVER_CONFIG_PATH", str(config_file))
148 |
149 | # Clear any email env vars
150 | for key in list(os.environ.keys()):
151 | if key.startswith("MCP_EMAIL_SERVER_") and "CONFIG_PATH" not in key:
152 | monkeypatch.delenv(key, raising=False)
153 |
154 | settings = Settings()
155 | assert len(settings.emails) == 0
156 |
157 |
158 | def test_settings_init_add_new_account(monkeypatch, tmp_path):
159 | """Test adding new account from env - covers lines 225-226."""
160 | config_file = tmp_path / "empty.toml"
161 | config_file.write_text("")
162 | monkeypatch.setenv("MCP_EMAIL_SERVER_CONFIG_PATH", str(config_file))
163 |
164 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
165 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "newpass")
166 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.new.com")
167 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.new.com")
168 | monkeypatch.setenv("MCP_EMAIL_SERVER_ACCOUNT_NAME", "newaccount")
169 |
170 | settings = Settings()
171 | assert len(settings.emails) == 1
172 | assert settings.emails[0].account_name == "newaccount"
173 |
174 |
175 | def test_settings_init_override_existing(monkeypatch, tmp_path):
176 | """Test overriding existing TOML account - covers lines 214-222."""
177 | config_file = tmp_path / "config.toml"
178 | config_file.write_text("""
179 | [[emails]]
180 | account_name = "existing"
181 | full_name = "Old Name"
182 | email_address = "[email protected]"
183 | created_at = "2025-01-01T00:00:00"
184 | updated_at = "2025-01-01T00:00:00"
185 |
186 | [emails.incoming]
187 | user_name = "olduser"
188 | password = "oldpass"
189 | host = "imap.old.com"
190 | port = 993
191 | use_ssl = true
192 |
193 | [emails.outgoing]
194 | user_name = "olduser"
195 | password = "oldpass"
196 | host = "smtp.old.com"
197 | port = 465
198 | use_ssl = true
199 | """)
200 |
201 | monkeypatch.setenv("MCP_EMAIL_SERVER_CONFIG_PATH", str(config_file))
202 | monkeypatch.setenv("MCP_EMAIL_SERVER_ACCOUNT_NAME", "existing")
203 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
204 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "newpass")
205 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.new.com")
206 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.new.com")
207 |
208 | settings = Settings()
209 | assert len(settings.emails) == 1
210 | assert settings.emails[0].account_name == "existing"
211 | assert settings.emails[0].email_address == "[email protected]" # Overridden
212 |
213 |
214 | def test_settings_init_loop_through_multiple_accounts(monkeypatch, tmp_path):
215 | """Test loop iteration with multiple accounts - covers lines 214-217."""
216 | config_file = tmp_path / "multi.toml"
217 | config_file.write_text("""
218 | [[emails]]
219 | account_name = "first"
220 | full_name = "First"
221 | email_address = "[email protected]"
222 | created_at = "2025-01-01T00:00:00"
223 | updated_at = "2025-01-01T00:00:00"
224 |
225 | [emails.incoming]
226 | user_name = "first"
227 | password = "pass1"
228 | host = "imap.first.com"
229 | port = 993
230 | use_ssl = true
231 |
232 | [emails.outgoing]
233 | user_name = "first"
234 | password = "pass1"
235 | host = "smtp.first.com"
236 | port = 465
237 | use_ssl = true
238 |
239 | [[emails]]
240 | account_name = "second"
241 | full_name = "Second"
242 | email_address = "[email protected]"
243 | created_at = "2025-01-01T00:00:00"
244 | updated_at = "2025-01-01T00:00:00"
245 |
246 | [emails.incoming]
247 | user_name = "second"
248 | password = "pass2"
249 | host = "imap.second.com"
250 | port = 993
251 | use_ssl = true
252 |
253 | [emails.outgoing]
254 | user_name = "second"
255 | password = "pass2"
256 | host = "smtp.second.com"
257 | port = 465
258 | use_ssl = true
259 |
260 | [[emails]]
261 | account_name = "third"
262 | full_name = "Third"
263 | email_address = "[email protected]"
264 | created_at = "2025-01-01T00:00:00"
265 | updated_at = "2025-01-01T00:00:00"
266 |
267 | [emails.incoming]
268 | user_name = "third"
269 | password = "pass3"
270 | host = "imap.third.com"
271 | port = 993
272 | use_ssl = true
273 |
274 | [emails.outgoing]
275 | user_name = "third"
276 | password = "pass3"
277 | host = "smtp.third.com"
278 | port = 465
279 | use_ssl = true
280 | """)
281 |
282 | monkeypatch.setenv("MCP_EMAIL_SERVER_CONFIG_PATH", str(config_file))
283 |
284 | # Override the third account (forces loop to iterate through all)
285 | monkeypatch.setenv("MCP_EMAIL_SERVER_ACCOUNT_NAME", "third")
286 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
287 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "envpass")
288 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.env.com")
289 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.env.com")
290 |
291 | settings = Settings()
292 | # Note: Our implementation replaces all TOML with env, so we only get 1 account
293 | assert len(settings.emails) == 1
294 | assert settings.emails[0].account_name == "third"
295 | assert settings.emails[0].email_address == "[email protected]"
296 |
297 |
298 | def test_email_settings_masked(monkeypatch):
299 | """Test the masked() method - covers line 182."""
300 | monkeypatch.setenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS", "[email protected]")
301 | monkeypatch.setenv("MCP_EMAIL_SERVER_PASSWORD", "secret123")
302 | monkeypatch.setenv("MCP_EMAIL_SERVER_IMAP_HOST", "imap.test.com")
303 | monkeypatch.setenv("MCP_EMAIL_SERVER_SMTP_HOST", "smtp.test.com")
304 |
305 | email = EmailSettings.from_env()
306 | assert email is not None
307 |
308 | masked = email.masked()
309 | assert masked.incoming.password == "********" # noqa: S105
310 | assert masked.outgoing.password == "********" # noqa: S105
311 | assert masked.email_address == "[email protected]"
312 |
```
--------------------------------------------------------------------------------
/mcp_email_server/config.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import datetime
4 | import os
5 | from pathlib import Path
6 | from typing import Any
7 |
8 | import tomli_w
9 | from pydantic import BaseModel, ConfigDict, Field, field_serializer, model_validator
10 | from pydantic_settings import (
11 | BaseSettings,
12 | PydanticBaseSettingsSource,
13 | SettingsConfigDict,
14 | TomlConfigSettingsSource,
15 | )
16 |
17 | from mcp_email_server.log import logger
18 |
19 | DEFAILT_CONFIG_PATH = "~/.config/zerolib/mcp_email_server/config.toml"
20 |
21 | CONFIG_PATH = Path(os.getenv("MCP_EMAIL_SERVER_CONFIG_PATH", DEFAILT_CONFIG_PATH)).expanduser().resolve()
22 |
23 |
24 | class EmailServer(BaseModel):
25 | user_name: str
26 | password: str
27 | host: str
28 | port: int
29 | use_ssl: bool = True # Usually port 465
30 | start_ssl: bool = False # Usually port 587
31 |
32 | def masked(self) -> EmailServer:
33 | return self.model_copy(update={"password": "********"})
34 |
35 |
36 | class AccountAttributes(BaseModel):
37 | model_config = ConfigDict(json_encoders={datetime.datetime: lambda v: v.isoformat()})
38 | account_name: str
39 | description: str = ""
40 | created_at: datetime.datetime = Field(default_factory=datetime.datetime.now)
41 | updated_at: datetime.datetime = Field(default_factory=datetime.datetime.now)
42 |
43 | @model_validator(mode="after")
44 | @classmethod
45 | def update_updated_at(cls, obj: AccountAttributes) -> AccountAttributes:
46 | """Update updated_at field."""
47 | # must disable validation to avoid infinite loop
48 | obj.model_config["validate_assignment"] = False
49 |
50 | # update updated_at field
51 | obj.updated_at = datetime.datetime.now()
52 |
53 | # enable validation again
54 | obj.model_config["validate_assignment"] = True
55 | return obj
56 |
57 | def __eq__(self, other: object) -> bool:
58 | if not isinstance(other, AccountAttributes):
59 | return NotImplemented
60 | return self.model_dump(exclude={"created_at", "updated_at"}) == other.model_dump(
61 | exclude={"created_at", "updated_at"}
62 | )
63 |
64 | @field_serializer("created_at", "updated_at")
65 | def serialize_datetime(self, v: datetime.datetime) -> str:
66 | return v.isoformat()
67 |
68 | def masked(self) -> AccountAttributes:
69 | return self.model_copy()
70 |
71 |
72 | class EmailSettings(AccountAttributes):
73 | full_name: str
74 | email_address: str
75 | incoming: EmailServer
76 | outgoing: EmailServer
77 |
78 | @classmethod
79 | def init(
80 | cls,
81 | *,
82 | account_name: str,
83 | full_name: str,
84 | email_address: str,
85 | user_name: str,
86 | password: str,
87 | imap_host: str,
88 | smtp_host: str,
89 | imap_user_name: str | None = None,
90 | imap_password: str | None = None,
91 | imap_port: int = 993,
92 | imap_ssl: bool = True,
93 | smtp_port: int = 465,
94 | smtp_ssl: bool = True,
95 | smtp_start_ssl: bool = False,
96 | smtp_user_name: str | None = None,
97 | smtp_password: str | None = None,
98 | ) -> EmailSettings:
99 | return cls(
100 | account_name=account_name,
101 | full_name=full_name,
102 | email_address=email_address,
103 | incoming=EmailServer(
104 | user_name=imap_user_name or user_name,
105 | password=imap_password or password,
106 | host=imap_host,
107 | port=imap_port,
108 | use_ssl=imap_ssl,
109 | ),
110 | outgoing=EmailServer(
111 | user_name=smtp_user_name or user_name,
112 | password=smtp_password or password,
113 | host=smtp_host,
114 | port=smtp_port,
115 | use_ssl=smtp_ssl,
116 | start_ssl=smtp_start_ssl,
117 | ),
118 | )
119 |
120 | @classmethod
121 | def from_env(cls) -> EmailSettings | None:
122 | """Create EmailSettings from environment variables.
123 |
124 | Expected environment variables:
125 | - MCP_EMAIL_SERVER_ACCOUNT_NAME (default: "default")
126 | - MCP_EMAIL_SERVER_FULL_NAME
127 | - MCP_EMAIL_SERVER_EMAIL_ADDRESS
128 | - MCP_EMAIL_SERVER_USER_NAME
129 | - MCP_EMAIL_SERVER_PASSWORD
130 | - MCP_EMAIL_SERVER_IMAP_HOST
131 | - MCP_EMAIL_SERVER_IMAP_PORT (default: 993)
132 | - MCP_EMAIL_SERVER_IMAP_SSL (default: true)
133 | - MCP_EMAIL_SERVER_SMTP_HOST
134 | - MCP_EMAIL_SERVER_SMTP_PORT (default: 465)
135 | - MCP_EMAIL_SERVER_SMTP_SSL (default: true)
136 | - MCP_EMAIL_SERVER_SMTP_START_SSL (default: false)
137 | """
138 | # Check if minimum required environment variables are set
139 | email_address = os.getenv("MCP_EMAIL_SERVER_EMAIL_ADDRESS")
140 | password = os.getenv("MCP_EMAIL_SERVER_PASSWORD")
141 |
142 | if not email_address or not password:
143 | return None
144 |
145 | # Parse boolean values
146 | def parse_bool(value: str | None, default: bool = True) -> bool:
147 | if value is None:
148 | return default
149 | return value.lower() in ("true", "1", "yes", "on")
150 |
151 | # Get all environment variables with defaults
152 | account_name = os.getenv("MCP_EMAIL_SERVER_ACCOUNT_NAME", "default")
153 | full_name = os.getenv("MCP_EMAIL_SERVER_FULL_NAME", email_address.split("@")[0])
154 | user_name = os.getenv("MCP_EMAIL_SERVER_USER_NAME", email_address)
155 | imap_host = os.getenv("MCP_EMAIL_SERVER_IMAP_HOST")
156 | smtp_host = os.getenv("MCP_EMAIL_SERVER_SMTP_HOST")
157 |
158 | # Required fields check
159 | if not imap_host or not smtp_host:
160 | logger.warning("Missing required email configuration environment variables (IMAP_HOST or SMTP_HOST)")
161 | return None
162 |
163 | try:
164 | return cls.init(
165 | account_name=account_name,
166 | full_name=full_name,
167 | email_address=email_address,
168 | user_name=user_name,
169 | password=password,
170 | imap_host=imap_host,
171 | imap_port=int(os.getenv("MCP_EMAIL_SERVER_IMAP_PORT", "993")),
172 | imap_ssl=parse_bool(os.getenv("MCP_EMAIL_SERVER_IMAP_SSL"), True),
173 | smtp_host=smtp_host,
174 | smtp_port=int(os.getenv("MCP_EMAIL_SERVER_SMTP_PORT", "465")),
175 | smtp_ssl=parse_bool(os.getenv("MCP_EMAIL_SERVER_SMTP_SSL"), True),
176 | smtp_start_ssl=parse_bool(os.getenv("MCP_EMAIL_SERVER_SMTP_START_SSL"), False),
177 | smtp_user_name=os.getenv("MCP_EMAIL_SERVER_SMTP_USER_NAME", user_name),
178 | smtp_password=os.getenv("MCP_EMAIL_SERVER_SMTP_PASSWORD", password),
179 | imap_user_name=os.getenv("MCP_EMAIL_SERVER_IMAP_USER_NAME", user_name),
180 | imap_password=os.getenv("MCP_EMAIL_SERVER_IMAP_PASSWORD", password),
181 | )
182 | except (ValueError, TypeError) as e:
183 | logger.error(f"Failed to create email settings from environment variables: {e}")
184 | return None
185 |
186 | def masked(self) -> EmailSettings:
187 | return self.model_copy(
188 | update={
189 | "incoming": self.incoming.masked(),
190 | "outgoing": self.outgoing.masked(),
191 | }
192 | )
193 |
194 |
195 | class ProviderSettings(AccountAttributes):
196 | provider_name: str
197 | api_key: str
198 |
199 | def masked(self) -> AccountAttributes:
200 | return self.model_copy(update={"api_key": "********"})
201 |
202 |
203 | class Settings(BaseSettings):
204 | emails: list[EmailSettings] = []
205 | providers: list[ProviderSettings] = []
206 | db_location: str = CONFIG_PATH.with_name("db.sqlite3").as_posix()
207 |
208 | model_config = SettingsConfigDict(toml_file=CONFIG_PATH, validate_assignment=True, revalidate_instances="always")
209 |
210 | def __init__(self, **data: Any) -> None:
211 | """Initialize Settings with support for environment variables."""
212 | super().__init__(**data)
213 |
214 | # Check for email configuration from environment variables
215 | env_email = EmailSettings.from_env()
216 | if env_email:
217 | # Check if this account already exists (from TOML)
218 | existing_account = None
219 | for i, email in enumerate(self.emails):
220 | if email.account_name == env_email.account_name:
221 | existing_account = i
222 | break
223 |
224 | if existing_account is not None:
225 | # Replace existing account with env configuration
226 | self.emails[existing_account] = env_email
227 | logger.info(f"Overriding email account '{env_email.account_name}' with environment variables")
228 | else:
229 | # Add new account from env
230 | self.emails.insert(0, env_email)
231 | logger.info(f"Added email account '{env_email.account_name}' from environment variables")
232 |
233 | def add_email(self, email: EmailSettings) -> None:
234 | """Use re-assigned for validation to work."""
235 | self.emails = [email, *self.emails]
236 |
237 | def add_provider(self, provider: ProviderSettings) -> None:
238 | """Use re-assigned for validation to work."""
239 | self.providers = [provider, *self.providers]
240 |
241 | def delete_email(self, account_name: str) -> None:
242 | """Use re-assigned for validation to work."""
243 | self.emails = [email for email in self.emails if email.account_name != account_name]
244 |
245 | def delete_provider(self, account_name: str) -> None:
246 | """Use re-assigned for validation to work."""
247 | self.providers = [provider for provider in self.providers if provider.account_name != account_name]
248 |
249 | def get_account(self, account_name: str, masked: bool = False) -> EmailSettings | ProviderSettings | None:
250 | for email in self.emails:
251 | if email.account_name == account_name:
252 | return email if not masked else email.masked()
253 | for provider in self.providers:
254 | if provider.account_name == account_name:
255 | return provider if not masked else provider.masked()
256 | return None
257 |
258 | def get_accounts(self, masked: bool = False) -> list[EmailSettings | ProviderSettings]:
259 | accounts = self.emails + self.providers
260 | if masked:
261 | return [account.masked() for account in accounts]
262 | return accounts
263 |
264 | @model_validator(mode="after")
265 | @classmethod
266 | def check_unique_account_names(cls, obj: Settings) -> Settings:
267 | account_names = set()
268 | for email in obj.emails:
269 | if email.account_name in account_names:
270 | raise ValueError(f"Duplicate account name {email.account_name}")
271 | account_names.add(email.account_name)
272 | for provider in obj.providers:
273 | if provider.account_name in account_names:
274 | raise ValueError(f"Duplicate account name {provider.account_name}")
275 | account_names.add(provider.account_name)
276 |
277 | return obj
278 |
279 | @classmethod
280 | def settings_customise_sources(
281 | cls,
282 | settings_cls: type[BaseSettings],
283 | init_settings: PydanticBaseSettingsSource,
284 | env_settings: PydanticBaseSettingsSource,
285 | dotenv_settings: PydanticBaseSettingsSource,
286 | file_secret_settings: PydanticBaseSettingsSource,
287 | ) -> tuple[PydanticBaseSettingsSource, ...]:
288 | return (TomlConfigSettingsSource(settings_cls),)
289 |
290 | def _to_toml(self) -> str:
291 | data = self.model_dump()
292 | return tomli_w.dumps(data)
293 |
294 | def store(self) -> None:
295 | toml_file = self.model_config["toml_file"]
296 | toml_file.parent.mkdir(parents=True, exist_ok=True)
297 | toml_file.write_text(self._to_toml())
298 | logger.info(f"Settings stored in {toml_file}")
299 |
300 |
301 | _settings = None
302 |
303 |
304 | def get_settings(reload: bool = False) -> Settings:
305 | global _settings
306 | if not _settings or reload:
307 | logger.info(f"Loading settings from {CONFIG_PATH}")
308 | _settings = Settings()
309 | return _settings
310 |
311 |
312 | def store_settings(settings: Settings | None = None) -> None:
313 | if not settings:
314 | settings = get_settings()
315 | settings.store()
316 | return
317 |
318 |
319 | def delete_settings() -> None:
320 | if not CONFIG_PATH.exists():
321 | logger.info(f"Settings file {CONFIG_PATH} does not exist")
322 | return
323 | CONFIG_PATH.unlink()
324 | logger.info(f"Deleted settings file {CONFIG_PATH}")
325 |
```
--------------------------------------------------------------------------------
/mcp_email_server/ui.py:
--------------------------------------------------------------------------------
```python
1 | import gradio as gr
2 |
3 | from mcp_email_server.config import EmailSettings, get_settings, store_settings
4 | from mcp_email_server.tools.installer import install_claude_desktop, is_installed, need_update, uninstall_claude_desktop
5 |
6 |
7 | def create_ui(): # noqa: C901
8 | # Create a Gradio interface
9 | with gr.Blocks(title="Email Settings Configuration") as app:
10 | gr.Markdown("# Email Settings Configuration")
11 |
12 | # Function to get current accounts
13 | def get_current_accounts():
14 | settings = get_settings(reload=True)
15 | email_accounts = [email.account_name for email in settings.emails]
16 | return email_accounts
17 |
18 | # Function to update account list display
19 | def update_account_list():
20 | settings = get_settings(reload=True)
21 | email_accounts = [email.account_name for email in settings.emails]
22 |
23 | if email_accounts:
24 | # Create a detailed list of accounts with more information
25 | accounts_details = []
26 | for email in settings.emails:
27 | details = [
28 | f"**Account Name:** {email.account_name}",
29 | f"**Full Name:** {email.full_name}",
30 | f"**Email Address:** {email.email_address}",
31 | ]
32 |
33 | if hasattr(email, "description") and email.description:
34 | details.append(f"**Description:** {email.description}")
35 |
36 | # Add IMAP/SMTP provider info if available
37 | if hasattr(email, "incoming") and hasattr(email.incoming, "host"):
38 | details.append(f"**IMAP Provider:** {email.incoming.host}")
39 |
40 | if hasattr(email, "outgoing") and hasattr(email.outgoing, "host"):
41 | details.append(f"**SMTP Provider:** {email.outgoing.host}")
42 |
43 | accounts_details.append("### " + email.account_name + "\n" + "\n".join(details) + "\n")
44 |
45 | accounts_md = "\n".join(accounts_details)
46 | return (
47 | f"## Configured Accounts\n{accounts_md}",
48 | gr.update(choices=email_accounts, value=None),
49 | gr.update(visible=True),
50 | )
51 | else:
52 | return (
53 | "No email accounts configured yet.",
54 | gr.update(choices=[], value=None),
55 | gr.update(visible=False),
56 | )
57 |
58 | # Display current email accounts and allow deletion
59 | with gr.Accordion("Current Email Accounts", open=True):
60 | # Display the list of accounts
61 | accounts_display = gr.Markdown("")
62 |
63 | # Create a dropdown to select account to delete
64 | account_to_delete = gr.Dropdown(choices=[], label="Select Account to Delete", interactive=True)
65 |
66 | # Status message for deletion
67 | delete_status = gr.Markdown("")
68 |
69 | # Delete button
70 | delete_btn = gr.Button("Delete Selected Account")
71 |
72 | # Function to delete an account
73 | def delete_email_account(account_name):
74 | if not account_name:
75 | return "Error: Please select an account to delete.", *update_account_list()
76 |
77 | try:
78 | # Get current settings
79 | settings = get_settings()
80 |
81 | # Delete the account
82 | settings.delete_email(account_name)
83 |
84 | # Store settings
85 | store_settings(settings)
86 |
87 | # Return success message and update the UI
88 | return f"Success: Email account '{account_name}' has been deleted.", *update_account_list()
89 | except Exception as e:
90 | return f"Error: {e!s}", *update_account_list()
91 |
92 | # Connect the delete button to the delete function
93 | delete_btn.click(
94 | fn=delete_email_account,
95 | inputs=[account_to_delete],
96 | outputs=[delete_status, accounts_display, account_to_delete, delete_btn],
97 | )
98 |
99 | # Initialize the account list
100 | app.load(
101 | fn=update_account_list,
102 | inputs=None,
103 | outputs=[accounts_display, account_to_delete, delete_btn],
104 | )
105 |
106 | # Form for adding a new email account
107 | with gr.Accordion("Add New Email Account", open=True):
108 | gr.Markdown("### Add New Email Account")
109 |
110 | # Basic account information
111 | account_name = gr.Textbox(label="Account Name", placeholder="e.g. work_email")
112 | full_name = gr.Textbox(label="Full Name", placeholder="e.g. John Doe")
113 | email_address = gr.Textbox(label="Email Address", placeholder="e.g. [email protected]")
114 |
115 | # Credentials
116 | user_name = gr.Textbox(label="Username", placeholder="e.g. [email protected]")
117 | password = gr.Textbox(label="Password", type="password")
118 |
119 | # IMAP settings
120 | with gr.Row():
121 | with gr.Column():
122 | gr.Markdown("### IMAP Settings")
123 | imap_host = gr.Textbox(label="IMAP Host", placeholder="e.g. imap.example.com")
124 | imap_port = gr.Number(label="IMAP Port", value=993)
125 | imap_ssl = gr.Checkbox(label="Use SSL", value=True)
126 | imap_user_name = gr.Textbox(
127 | label="IMAP Username (optional)", placeholder="Leave empty to use the same as above"
128 | )
129 | imap_password = gr.Textbox(
130 | label="IMAP Password (optional)",
131 | type="password",
132 | placeholder="Leave empty to use the same as above",
133 | )
134 |
135 | # SMTP settings
136 | with gr.Column():
137 | gr.Markdown("### SMTP Settings")
138 | smtp_host = gr.Textbox(label="SMTP Host", placeholder="e.g. smtp.example.com")
139 | smtp_port = gr.Number(label="SMTP Port", value=465)
140 | smtp_ssl = gr.Checkbox(label="Use SSL", value=True)
141 | smtp_start_ssl = gr.Checkbox(label="Start SSL", value=False)
142 | smtp_user_name = gr.Textbox(
143 | label="SMTP Username (optional)", placeholder="Leave empty to use the same as above"
144 | )
145 | smtp_password = gr.Textbox(
146 | label="SMTP Password (optional)",
147 | type="password",
148 | placeholder="Leave empty to use the same as above",
149 | )
150 |
151 | # Status message
152 | status_message = gr.Markdown("")
153 |
154 | # Save button
155 | save_btn = gr.Button("Save Email Settings")
156 |
157 | # Function to save settings
158 | def save_email_settings(
159 | account_name,
160 | full_name,
161 | email_address,
162 | user_name,
163 | password,
164 | imap_host,
165 | imap_port,
166 | imap_ssl,
167 | imap_user_name,
168 | imap_password,
169 | smtp_host,
170 | smtp_port,
171 | smtp_ssl,
172 | smtp_start_ssl,
173 | smtp_user_name,
174 | smtp_password,
175 | ):
176 | try:
177 | # Validate required fields
178 | if not account_name or not full_name or not email_address or not user_name or not password:
179 | # Get account list update
180 | account_md, account_choices, btn_visible = update_account_list()
181 | return (
182 | "Error: Please fill in all required fields.",
183 | account_md,
184 | account_choices,
185 | btn_visible,
186 | account_name,
187 | full_name,
188 | email_address,
189 | user_name,
190 | password,
191 | imap_host,
192 | imap_port,
193 | imap_ssl,
194 | imap_user_name,
195 | imap_password,
196 | smtp_host,
197 | smtp_port,
198 | smtp_ssl,
199 | smtp_start_ssl,
200 | smtp_user_name,
201 | smtp_password,
202 | )
203 |
204 | if not imap_host or not smtp_host:
205 | # Get account list update
206 | account_md, account_choices, btn_visible = update_account_list()
207 | return (
208 | "Error: IMAP and SMTP hosts are required.",
209 | account_md,
210 | account_choices,
211 | btn_visible,
212 | account_name,
213 | full_name,
214 | email_address,
215 | user_name,
216 | password,
217 | imap_host,
218 | imap_port,
219 | imap_ssl,
220 | imap_user_name,
221 | imap_password,
222 | smtp_host,
223 | smtp_port,
224 | smtp_ssl,
225 | smtp_start_ssl,
226 | smtp_user_name,
227 | smtp_password,
228 | )
229 |
230 | # Get current settings
231 | settings = get_settings()
232 |
233 | # Check if account name already exists
234 | for email in settings.emails:
235 | if email.account_name == account_name:
236 | # Get account list update
237 | account_md, account_choices, btn_visible = update_account_list()
238 | return (
239 | f"Error: Account name '{account_name}' already exists.",
240 | account_md,
241 | account_choices,
242 | btn_visible,
243 | account_name,
244 | full_name,
245 | email_address,
246 | user_name,
247 | password,
248 | imap_host,
249 | imap_port,
250 | imap_ssl,
251 | imap_user_name,
252 | imap_password,
253 | smtp_host,
254 | smtp_port,
255 | smtp_ssl,
256 | smtp_start_ssl,
257 | smtp_user_name,
258 | smtp_password,
259 | )
260 |
261 | # Create new email settings
262 | email_settings = EmailSettings.init(
263 | account_name=account_name,
264 | full_name=full_name,
265 | email_address=email_address,
266 | user_name=user_name,
267 | password=password,
268 | imap_host=imap_host,
269 | smtp_host=smtp_host,
270 | imap_port=int(imap_port),
271 | imap_ssl=imap_ssl,
272 | smtp_port=int(smtp_port),
273 | smtp_ssl=smtp_ssl,
274 | smtp_start_ssl=smtp_start_ssl,
275 | imap_user_name=imap_user_name if imap_user_name else None,
276 | imap_password=imap_password if imap_password else None,
277 | smtp_user_name=smtp_user_name if smtp_user_name else None,
278 | smtp_password=smtp_password if smtp_password else None,
279 | )
280 |
281 | # Add to settings
282 | settings.add_email(email_settings)
283 |
284 | # Store settings
285 | store_settings(settings)
286 |
287 | # Get account list update
288 | account_md, account_choices, btn_visible = update_account_list()
289 |
290 | # Return success message, update the UI, and clear form fields
291 | return (
292 | f"Success: Email account '{account_name}' has been added.",
293 | account_md,
294 | account_choices,
295 | btn_visible,
296 | "", # Clear account_name
297 | "", # Clear full_name
298 | "", # Clear email_address
299 | "", # Clear user_name
300 | "", # Clear password
301 | "", # Clear imap_host
302 | 993, # Reset imap_port
303 | True, # Reset imap_ssl
304 | "", # Clear imap_user_name
305 | "", # Clear imap_password
306 | "", # Clear smtp_host
307 | 465, # Reset smtp_port
308 | True, # Reset smtp_ssl
309 | False, # Reset smtp_start_ssl
310 | "", # Clear smtp_user_name
311 | "", # Clear smtp_password
312 | )
313 | except Exception as e:
314 | # Get account list update
315 | account_md, account_choices, btn_visible = update_account_list()
316 | return (
317 | f"Error: {e!s}",
318 | account_md,
319 | account_choices,
320 | btn_visible,
321 | account_name,
322 | full_name,
323 | email_address,
324 | user_name,
325 | password,
326 | imap_host,
327 | imap_port,
328 | imap_ssl,
329 | imap_user_name,
330 | imap_password,
331 | smtp_host,
332 | smtp_port,
333 | smtp_ssl,
334 | smtp_start_ssl,
335 | smtp_user_name,
336 | smtp_password,
337 | )
338 |
339 | # Connect the save button to the save function
340 | save_btn.click(
341 | fn=save_email_settings,
342 | inputs=[
343 | account_name,
344 | full_name,
345 | email_address,
346 | user_name,
347 | password,
348 | imap_host,
349 | imap_port,
350 | imap_ssl,
351 | imap_user_name,
352 | imap_password,
353 | smtp_host,
354 | smtp_port,
355 | smtp_ssl,
356 | smtp_start_ssl,
357 | smtp_user_name,
358 | smtp_password,
359 | ],
360 | outputs=[
361 | status_message,
362 | accounts_display,
363 | account_to_delete,
364 | delete_btn,
365 | account_name,
366 | full_name,
367 | email_address,
368 | user_name,
369 | password,
370 | imap_host,
371 | imap_port,
372 | imap_ssl,
373 | imap_user_name,
374 | imap_password,
375 | smtp_host,
376 | smtp_port,
377 | smtp_ssl,
378 | smtp_start_ssl,
379 | smtp_user_name,
380 | smtp_password,
381 | ],
382 | )
383 |
384 | # Claude Desktop Integration
385 | with gr.Accordion("Claude Desktop Integration", open=True):
386 | gr.Markdown("### Claude Desktop Integration")
387 |
388 | # Status display for Claude Desktop integration
389 | claude_status = gr.Markdown("")
390 |
391 | # Function to check and update Claude Desktop status
392 | def update_claude_status():
393 | if is_installed():
394 | if need_update():
395 | return "Claude Desktop integration is installed but needs to be updated."
396 | else:
397 | return "Claude Desktop integration is installed and up to date."
398 | else:
399 | return "Claude Desktop integration is not installed."
400 |
401 | # Buttons for Claude Desktop actions
402 | with gr.Row():
403 | install_update_btn = gr.Button("Install to Claude Desktop")
404 | uninstall_btn = gr.Button("Uninstall from Claude Desktop")
405 |
406 | # Functions for Claude Desktop actions
407 | def install_or_update_claude():
408 | try:
409 | install_claude_desktop()
410 | status = update_claude_status()
411 | # Update button states based on new status
412 | is_inst = is_installed()
413 | needs_upd = need_update()
414 |
415 | button_text = "Update Claude Desktop" if (is_inst and needs_upd) else "Install to Claude Desktop"
416 | button_interactive = not (is_inst and not needs_upd)
417 |
418 | return [
419 | status,
420 | gr.update(value=button_text, interactive=button_interactive),
421 | gr.update(interactive=is_inst),
422 | ]
423 | except Exception as e:
424 | return [f"Error installing/updating Claude Desktop: {e!s}", gr.update(), gr.update()]
425 |
426 | def uninstall_from_claude():
427 | try:
428 | uninstall_claude_desktop()
429 | status = update_claude_status()
430 | # Update button states based on new status
431 | is_inst = is_installed()
432 | needs_upd = need_update()
433 |
434 | button_text = "Update Claude Desktop" if (is_inst and needs_upd) else "Install to Claude Desktop"
435 | button_interactive = not (is_inst and not needs_upd)
436 |
437 | return [
438 | status,
439 | gr.update(value=button_text, interactive=button_interactive),
440 | gr.update(interactive=is_inst),
441 | ]
442 | except Exception as e:
443 | return [f"Error uninstalling from Claude Desktop: {e!s}", gr.update(), gr.update()]
444 |
445 | # Function to update button states based on installation status
446 | def update_button_states():
447 | status = update_claude_status()
448 | is_inst = is_installed()
449 | needs_upd = need_update()
450 |
451 | button_text = "Update Claude Desktop" if (is_inst and needs_upd) else "Install to Claude Desktop"
452 | button_interactive = not (is_inst and not needs_upd)
453 |
454 | return [
455 | status,
456 | gr.update(value=button_text, interactive=button_interactive),
457 | gr.update(interactive=is_inst),
458 | ]
459 |
460 | # Connect buttons to functions
461 | install_update_btn.click(
462 | fn=install_or_update_claude, inputs=[], outputs=[claude_status, install_update_btn, uninstall_btn]
463 | )
464 | uninstall_btn.click(
465 | fn=uninstall_from_claude, inputs=[], outputs=[claude_status, install_update_btn, uninstall_btn]
466 | )
467 |
468 | # Initialize Claude Desktop status and button states
469 | app.load(fn=update_button_states, inputs=None, outputs=[claude_status, install_update_btn, uninstall_btn])
470 |
471 | return app
472 |
473 |
474 | def main():
475 | app = create_ui()
476 | app.launch(inbrowser=True)
477 |
478 |
479 | if __name__ == "__main__":
480 | main()
481 |
```
--------------------------------------------------------------------------------
/mcp_email_server/emails/classic.py:
--------------------------------------------------------------------------------
```python
1 | import email.utils
2 | from collections.abc import AsyncGenerator
3 | from datetime import datetime
4 | from email.header import Header
5 | from email.mime.text import MIMEText
6 | from email.parser import BytesParser
7 | from email.policy import default
8 | from typing import Any
9 |
10 | import aioimaplib
11 | import aiosmtplib
12 |
13 | from mcp_email_server.config import EmailServer, EmailSettings
14 | from mcp_email_server.emails import EmailHandler
15 | from mcp_email_server.emails.models import (
16 | EmailBodyResponse,
17 | EmailContentBatchResponse,
18 | EmailMetadata,
19 | EmailMetadataPageResponse,
20 | )
21 | from mcp_email_server.log import logger
22 |
23 |
24 | class EmailClient:
25 | def __init__(self, email_server: EmailServer, sender: str | None = None):
26 | self.email_server = email_server
27 | self.sender = sender or email_server.user_name
28 |
29 | self.imap_class = aioimaplib.IMAP4_SSL if self.email_server.use_ssl else aioimaplib.IMAP4
30 |
31 | self.smtp_use_tls = self.email_server.use_ssl
32 | self.smtp_start_tls = self.email_server.start_ssl
33 |
34 | def _parse_email_data(self, raw_email: bytes, email_id: str | None = None) -> dict[str, Any]: # noqa: C901
35 | """Parse raw email data into a structured dictionary."""
36 | parser = BytesParser(policy=default)
37 | email_message = parser.parsebytes(raw_email)
38 |
39 | # Extract email parts
40 | subject = email_message.get("Subject", "")
41 | sender = email_message.get("From", "")
42 | date_str = email_message.get("Date", "")
43 |
44 | # Extract recipients
45 | to_addresses = []
46 | to_header = email_message.get("To", "")
47 | if to_header:
48 | # Simple parsing - split by comma and strip whitespace
49 | to_addresses = [addr.strip() for addr in to_header.split(",")]
50 |
51 | # Also check CC recipients
52 | cc_header = email_message.get("Cc", "")
53 | if cc_header:
54 | to_addresses.extend([addr.strip() for addr in cc_header.split(",")])
55 |
56 | # Parse date
57 | try:
58 | date_tuple = email.utils.parsedate_tz(date_str)
59 | date = datetime.fromtimestamp(email.utils.mktime_tz(date_tuple)) if date_tuple else datetime.now()
60 | except Exception:
61 | date = datetime.now()
62 |
63 | # Get body content
64 | body = ""
65 | attachments = []
66 |
67 | if email_message.is_multipart():
68 | for part in email_message.walk():
69 | content_type = part.get_content_type()
70 | content_disposition = str(part.get("Content-Disposition", ""))
71 |
72 | # Handle attachments
73 | if "attachment" in content_disposition:
74 | filename = part.get_filename()
75 | if filename:
76 | attachments.append(filename)
77 | # Handle text parts
78 | elif content_type == "text/plain":
79 | body_part = part.get_payload(decode=True)
80 | if body_part:
81 | charset = part.get_content_charset("utf-8")
82 | try:
83 | body += body_part.decode(charset)
84 | except UnicodeDecodeError:
85 | body += body_part.decode("utf-8", errors="replace")
86 | else:
87 | # Handle plain text emails
88 | payload = email_message.get_payload(decode=True)
89 | if payload:
90 | charset = email_message.get_content_charset("utf-8")
91 | try:
92 | body = payload.decode(charset)
93 | except UnicodeDecodeError:
94 | body = payload.decode("utf-8", errors="replace")
95 | # TODO: Allow retrieving full email body
96 | if body and len(body) > 20000:
97 | body = body[:20000] + "...[TRUNCATED]"
98 | return {
99 | "email_id": email_id or "",
100 | "subject": subject,
101 | "from": sender,
102 | "to": to_addresses,
103 | "body": body,
104 | "date": date,
105 | "attachments": attachments,
106 | }
107 |
108 | @staticmethod
109 | def _build_search_criteria(
110 | before: datetime | None = None,
111 | since: datetime | None = None,
112 | subject: str | None = None,
113 | body: str | None = None,
114 | text: str | None = None,
115 | from_address: str | None = None,
116 | to_address: str | None = None,
117 | ):
118 | search_criteria = []
119 | if before:
120 | search_criteria.extend(["BEFORE", before.strftime("%d-%b-%Y").upper()])
121 | if since:
122 | search_criteria.extend(["SINCE", since.strftime("%d-%b-%Y").upper()])
123 | if subject:
124 | search_criteria.extend(["SUBJECT", subject])
125 | if body:
126 | search_criteria.extend(["BODY", body])
127 | if text:
128 | search_criteria.extend(["TEXT", text])
129 | if from_address:
130 | search_criteria.extend(["FROM", from_address])
131 | if to_address:
132 | search_criteria.extend(["TO", to_address])
133 |
134 | # If no specific criteria, search for ALL
135 | if not search_criteria:
136 | search_criteria = ["ALL"]
137 |
138 | return search_criteria
139 |
140 | async def get_email_count(
141 | self,
142 | before: datetime | None = None,
143 | since: datetime | None = None,
144 | subject: str | None = None,
145 | from_address: str | None = None,
146 | to_address: str | None = None,
147 | ) -> int:
148 | imap = self.imap_class(self.email_server.host, self.email_server.port)
149 | try:
150 | # Wait for the connection to be established
151 | await imap._client_task
152 | await imap.wait_hello_from_server()
153 |
154 | # Login and select inbox
155 | await imap.login(self.email_server.user_name, self.email_server.password)
156 | await imap.select("INBOX")
157 | search_criteria = self._build_search_criteria(
158 | before, since, subject, from_address=from_address, to_address=to_address
159 | )
160 | logger.info(f"Count: Search criteria: {search_criteria}")
161 | # Search for messages and count them - use UID SEARCH for consistency
162 | _, messages = await imap.uid_search(*search_criteria)
163 | return len(messages[0].split())
164 | finally:
165 | # Ensure we logout properly
166 | try:
167 | await imap.logout()
168 | except Exception as e:
169 | logger.info(f"Error during logout: {e}")
170 |
171 | async def get_emails_metadata_stream( # noqa: C901
172 | self,
173 | page: int = 1,
174 | page_size: int = 10,
175 | before: datetime | None = None,
176 | since: datetime | None = None,
177 | subject: str | None = None,
178 | from_address: str | None = None,
179 | to_address: str | None = None,
180 | order: str = "desc",
181 | ) -> AsyncGenerator[dict[str, Any], None]:
182 | imap = self.imap_class(self.email_server.host, self.email_server.port)
183 | try:
184 | # Wait for the connection to be established
185 | await imap._client_task
186 | await imap.wait_hello_from_server()
187 |
188 | # Login and select inbox
189 | await imap.login(self.email_server.user_name, self.email_server.password)
190 | try:
191 | await imap.id(name="mcp-email-server", version="1.0.0")
192 | except Exception as e:
193 | logger.warning(f"IMAP ID command failed: {e!s}")
194 | await imap.select("INBOX")
195 |
196 | search_criteria = self._build_search_criteria(
197 | before, since, subject, from_address=from_address, to_address=to_address
198 | )
199 | logger.info(f"Get metadata: Search criteria: {search_criteria}")
200 |
201 | # Search for messages - use UID SEARCH for better compatibility
202 | _, messages = await imap.uid_search(*search_criteria)
203 |
204 | # Handle empty or None responses
205 | if not messages or not messages[0]:
206 | logger.warning("No messages returned from search")
207 | email_ids = []
208 | else:
209 | email_ids = messages[0].split()
210 | logger.info(f"Found {len(email_ids)} email IDs")
211 | start = (page - 1) * page_size
212 | end = start + page_size
213 |
214 | if order == "desc":
215 | email_ids.reverse()
216 |
217 | # Fetch each message's metadata only
218 | for _, email_id in enumerate(email_ids[start:end]):
219 | try:
220 | # Convert email_id from bytes to string
221 | email_id_str = email_id.decode("utf-8")
222 |
223 | # Fetch only headers to get metadata without body
224 | _, data = await imap.uid("fetch", email_id_str, "BODY.PEEK[HEADER]")
225 |
226 | if not data:
227 | logger.error(f"Failed to fetch headers for UID {email_id_str}")
228 | continue
229 |
230 | # Find the email headers in the response
231 | raw_headers = None
232 | if len(data) > 1 and isinstance(data[1], bytearray):
233 | raw_headers = bytes(data[1])
234 | else:
235 | # Search through all items for header content
236 | for item in data:
237 | if isinstance(item, bytes | bytearray) and len(item) > 10:
238 | # Skip IMAP protocol responses
239 | if isinstance(item, bytes) and b"FETCH" in item:
240 | continue
241 | # This is likely the header content
242 | raw_headers = bytes(item) if isinstance(item, bytearray) else item
243 | break
244 |
245 | if raw_headers:
246 | try:
247 | # Parse headers only
248 | parser = BytesParser(policy=default)
249 | email_message = parser.parsebytes(raw_headers)
250 |
251 | # Extract metadata
252 | subject = email_message.get("Subject", "")
253 | sender = email_message.get("From", "")
254 | date_str = email_message.get("Date", "")
255 |
256 | # Extract recipients
257 | to_addresses = []
258 | to_header = email_message.get("To", "")
259 | if to_header:
260 | to_addresses = [addr.strip() for addr in to_header.split(",")]
261 |
262 | cc_header = email_message.get("Cc", "")
263 | if cc_header:
264 | to_addresses.extend([addr.strip() for addr in cc_header.split(",")])
265 |
266 | # Parse date
267 | try:
268 | date_tuple = email.utils.parsedate_tz(date_str)
269 | date = (
270 | datetime.fromtimestamp(email.utils.mktime_tz(date_tuple))
271 | if date_tuple
272 | else datetime.now()
273 | )
274 | except Exception:
275 | date = datetime.now()
276 |
277 | # For metadata, we don't fetch attachments to save bandwidth
278 | # We'll mark it as unknown for now
279 | metadata = {
280 | "email_id": email_id_str,
281 | "subject": subject,
282 | "from": sender,
283 | "to": to_addresses,
284 | "date": date,
285 | "attachments": [], # We don't fetch attachment info for metadata
286 | }
287 | yield metadata
288 | except Exception as e:
289 | # Log error but continue with other emails
290 | logger.error(f"Error parsing email metadata: {e!s}")
291 | else:
292 | logger.error(f"Could not find header data in response for email ID: {email_id_str}")
293 | except Exception as e:
294 | logger.error(f"Error fetching email metadata {email_id}: {e!s}")
295 | finally:
296 | # Ensure we logout properly
297 | try:
298 | await imap.logout()
299 | except Exception as e:
300 | logger.info(f"Error during logout: {e}")
301 |
302 | def _check_email_content(self, data: list) -> bool:
303 | """Check if the fetched data contains actual email content."""
304 | for item in data:
305 | if isinstance(item, bytes) and b"FETCH (" in item and b"RFC822" not in item and b"BODY" not in item:
306 | # This is just metadata, not actual content
307 | continue
308 | elif isinstance(item, bytes | bytearray) and len(item) > 100:
309 | # This looks like email content
310 | return True
311 | return False
312 |
313 | def _extract_raw_email(self, data: list) -> bytes | None:
314 | """Extract raw email bytes from IMAP response data."""
315 | # The email content is typically at index 1 as a bytearray
316 | if len(data) > 1 and isinstance(data[1], bytearray):
317 | return bytes(data[1])
318 |
319 | # Search through all items for email content
320 | for item in data:
321 | if isinstance(item, bytes | bytearray) and len(item) > 100:
322 | # Skip IMAP protocol responses
323 | if isinstance(item, bytes) and b"FETCH" in item:
324 | continue
325 | # This is likely the email content
326 | return bytes(item) if isinstance(item, bytearray) else item
327 | return None
328 |
329 | async def _fetch_email_with_formats(self, imap, email_id: str) -> list | None:
330 | """Try different fetch formats to get email data."""
331 | fetch_formats = ["RFC822", "BODY[]", "BODY.PEEK[]", "(BODY.PEEK[])"]
332 |
333 | for fetch_format in fetch_formats:
334 | try:
335 | _, data = await imap.uid("fetch", email_id, fetch_format)
336 |
337 | if data and len(data) > 0 and self._check_email_content(data):
338 | return data
339 |
340 | except Exception as e:
341 | logger.debug(f"Fetch format {fetch_format} failed: {e}")
342 |
343 | return None
344 |
345 | async def get_email_body_by_id(self, email_id: str) -> dict[str, Any] | None:
346 | imap = self.imap_class(self.email_server.host, self.email_server.port)
347 | try:
348 | # Wait for the connection to be established
349 | await imap._client_task
350 | await imap.wait_hello_from_server()
351 |
352 | # Login and select inbox
353 | await imap.login(self.email_server.user_name, self.email_server.password)
354 | try:
355 | await imap.id(name="mcp-email-server", version="1.0.0")
356 | except Exception as e:
357 | logger.warning(f"IMAP ID command failed: {e!s}")
358 | await imap.select("INBOX")
359 |
360 | # Fetch the specific email by UID
361 | data = await self._fetch_email_with_formats(imap, email_id)
362 | if not data:
363 | logger.error(f"Failed to fetch UID {email_id} with any format")
364 | return None
365 |
366 | # Extract raw email data
367 | raw_email = self._extract_raw_email(data)
368 | if not raw_email:
369 | logger.error(f"Could not find email data in response for email ID: {email_id}")
370 | return None
371 |
372 | # Parse the email
373 | try:
374 | return self._parse_email_data(raw_email, email_id)
375 | except Exception as e:
376 | logger.error(f"Error parsing email: {e!s}")
377 | return None
378 |
379 | finally:
380 | # Ensure we logout properly
381 | try:
382 | await imap.logout()
383 | except Exception as e:
384 | logger.info(f"Error during logout: {e}")
385 |
386 | async def send_email(
387 | self,
388 | recipients: list[str],
389 | subject: str,
390 | body: str,
391 | cc: list[str] | None = None,
392 | bcc: list[str] | None = None,
393 | html: bool = False,
394 | ):
395 | # Create message with UTF-8 encoding to support special characters
396 | content_type = "html" if html else "plain"
397 | msg = MIMEText(body, content_type, "utf-8")
398 |
399 | # Handle subject with special characters
400 | if any(ord(c) > 127 for c in subject):
401 | msg["Subject"] = Header(subject, "utf-8")
402 | else:
403 | msg["Subject"] = subject
404 |
405 | # Handle sender name with special characters
406 | if any(ord(c) > 127 for c in self.sender):
407 | msg["From"] = Header(self.sender, "utf-8")
408 | else:
409 | msg["From"] = self.sender
410 |
411 | msg["To"] = ", ".join(recipients)
412 |
413 | # Add CC header if provided (visible to recipients)
414 | if cc:
415 | msg["Cc"] = ", ".join(cc)
416 |
417 | # Note: BCC recipients are not added to headers (they remain hidden)
418 | # but will be included in the actual recipients for SMTP delivery
419 |
420 | async with aiosmtplib.SMTP(
421 | hostname=self.email_server.host,
422 | port=self.email_server.port,
423 | start_tls=self.smtp_start_tls,
424 | use_tls=self.smtp_use_tls,
425 | ) as smtp:
426 | await smtp.login(self.email_server.user_name, self.email_server.password)
427 |
428 | # Create a combined list of all recipients for delivery
429 | all_recipients = recipients.copy()
430 | if cc:
431 | all_recipients.extend(cc)
432 | if bcc:
433 | all_recipients.extend(bcc)
434 |
435 | await smtp.send_message(msg, recipients=all_recipients)
436 |
437 |
438 | class ClassicEmailHandler(EmailHandler):
439 | def __init__(self, email_settings: EmailSettings):
440 | self.email_settings = email_settings
441 | self.incoming_client = EmailClient(email_settings.incoming)
442 | self.outgoing_client = EmailClient(
443 | email_settings.outgoing,
444 | sender=f"{email_settings.full_name} <{email_settings.email_address}>",
445 | )
446 |
447 | async def get_emails_metadata(
448 | self,
449 | page: int = 1,
450 | page_size: int = 10,
451 | before: datetime | None = None,
452 | since: datetime | None = None,
453 | subject: str | None = None,
454 | from_address: str | None = None,
455 | to_address: str | None = None,
456 | order: str = "desc",
457 | ) -> EmailMetadataPageResponse:
458 | emails = []
459 | async for email_data in self.incoming_client.get_emails_metadata_stream(
460 | page, page_size, before, since, subject, from_address, to_address, order
461 | ):
462 | emails.append(EmailMetadata.from_email(email_data))
463 | total = await self.incoming_client.get_email_count(
464 | before, since, subject, from_address=from_address, to_address=to_address
465 | )
466 | return EmailMetadataPageResponse(
467 | page=page,
468 | page_size=page_size,
469 | before=before,
470 | since=since,
471 | subject=subject,
472 | emails=emails,
473 | total=total,
474 | )
475 |
476 | async def get_emails_content(self, email_ids: list[str]) -> EmailContentBatchResponse:
477 | """Batch retrieve email body content"""
478 | emails = []
479 | failed_ids = []
480 |
481 | for email_id in email_ids:
482 | try:
483 | email_data = await self.incoming_client.get_email_body_by_id(email_id)
484 | if email_data:
485 | emails.append(
486 | EmailBodyResponse(
487 | email_id=email_data["email_id"],
488 | subject=email_data["subject"],
489 | sender=email_data["from"],
490 | recipients=email_data["to"],
491 | date=email_data["date"],
492 | body=email_data["body"],
493 | attachments=email_data["attachments"],
494 | )
495 | )
496 | else:
497 | failed_ids.append(email_id)
498 | except Exception as e:
499 | logger.error(f"Failed to retrieve email {email_id}: {e}")
500 | failed_ids.append(email_id)
501 |
502 | return EmailContentBatchResponse(
503 | emails=emails,
504 | requested_count=len(email_ids),
505 | retrieved_count=len(emails),
506 | failed_ids=failed_ids,
507 | )
508 |
509 | async def send_email(
510 | self,
511 | recipients: list[str],
512 | subject: str,
513 | body: str,
514 | cc: list[str] | None = None,
515 | bcc: list[str] | None = None,
516 | html: bool = False,
517 | ) -> None:
518 | await self.outgoing_client.send_email(recipients, subject, body, cc, bcc, html)
519 |
```