Compare commits

..

146 Commits

Author SHA1 Message Date
Chris Coutinho 1dfdad5fad Update README, docstrings, and test scope for temporary_addressbook 2025-08-03 14:42:16 +02:00
Chris Coutinho 72cb62a101 test(contacts): Add unit/integration tests for a few tools 2025-08-03 14:36:16 +02:00
Chris Coutinho 21fc55320b Fix scoping 2025-08-03 14:25:01 +02:00
Chris Coutinho ad3e288203 test: Replace test_*_clients with single nc_client for tests 2025-08-03 14:22:45 +02:00
Chris Coutinho 0a97357a9c remove main.py 2025-08-03 14:17:29 +02:00
Chris Coutinho 70f01bf40a Add files 2025-08-03 14:16:55 +02:00
Chris Coutinho 37b1057d2a feat(contacts): Initialize Contacts App 2025-08-03 14:15:37 +02:00
Chris Coutinho ad95140416 Merge pull request #102 from cbcoutinho/renovate/docker-metadata-action-digest
chore(deps): update docker/metadata-action digest to c1e5197
2025-08-01 12:43:12 +02:00
github-actions[bot] 73fb56f73d bump: version 0.6.0 → 0.6.1 2025-08-01 10:41:12 +00:00
Chris Coutinho 9cc5300aa8 Merge pull request #96 from cbcoutinho/refactor/server
Refactor server tools and resources
2025-08-01 12:40:52 +02:00
Chris Coutinho be466abc0c Update README for deployment 2025-08-01 12:36:52 +02:00
Chris Coutinho 8956945e9d chore: sort imports 2025-08-01 12:21:32 +02:00
Chris Coutinho a9f3e1b00d Remove app check 2025-08-01 12:16:11 +02:00
Chris Coutinho a5e3f949c2 Use unique calendar_test_client 2025-08-01 12:08:27 +02:00
renovate-bot-cbcoutinho[bot] acc505aa01 chore(deps): update docker/metadata-action digest to c1e5197 2025-08-01 10:06:53 +00:00
Chris Coutinho 69fccb496a Use self._make_request 2025-08-01 11:05:28 +02:00
Chris Coutinho 6bdbb6ea6c Create sample calendar 2025-08-01 10:26:56 +02:00
Chris Coutinho 0b8a3aa646 Prepare calendar before running tests 2025-08-01 09:29:15 +02:00
Chris Coutinho ed270bb926 Add OCS-APIRequest: true to tables app check 2025-08-01 09:11:14 +02:00
Chris Coutinho 56e5298cce Wait for apps to be installed 2025-08-01 09:07:01 +02:00
Chris Coutinho 2bcfd3d7ee fix(calendar): Fix iCalendar date vs datetime format 2025-08-01 08:34:51 +02:00
Chris Coutinho 75235d6013 Refactor datetime 2025-07-31 14:51:33 +02:00
Chris Coutinho 19631838bb Merge remote-tracking branch 'origin/master' into refactor/server 2025-07-31 11:50:17 +02:00
Chris Coutinho 3cab343416 Merge pull request #99 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.4
2025-07-31 07:22:55 +02:00
renovate-bot-cbcoutinho[bot] 1a253af1c0 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.4 2025-07-30 22:06:16 +00:00
Chris Coutinho b81fe6dfa0 fix(calendar): Remove try/except in calendar API 2025-07-30 11:03:01 +02:00
Chris Coutinho 2a5b12343c chore: pre-commit 2025-07-29 15:13:02 +02:00
Chris Coutinho 66d306708d test(calendar): Enable calendar app in CICD 2025-07-29 15:12:39 +02:00
Chris Coutinho e7598a5467 format 2025-07-29 15:00:23 +02:00
Chris Coutinho fb6aa954b6 chore: ruff check 2025-07-29 09:11:25 +02:00
Chris Coutinho 02ad283a01 chore: format 2025-07-29 09:09:10 +02:00
Chris Coutinho 13ba9ef2e6 Merge remote-tracking branch 'origin/master' into refactor/server 2025-07-29 09:08:17 +02:00
github-actions[bot] 4767e88d2b bump: version 0.5.0 → 0.6.0 2025-07-29 05:40:28 +00:00
Chris Coutinho e38d0a8bdc Merge pull request #98 from cbcoutinho/renovate/nextcloud-31.0.7
chore(deps): update nextcloud:31.0.7 docker digest to 81dc361
2025-07-29 07:40:13 +02:00
Chris Coutinho 1dca929983 Merge pull request #95 from neovasky/master
feat(calendar): add comprehensive Calendar app support via CalDAV protocol
2025-07-29 07:40:02 +02:00
renovate-bot-cbcoutinho[bot] 6a2bd4d274 chore(deps): update nextcloud:31.0.7 docker digest to 81dc361 2025-07-29 04:11:46 +00:00
Neovasky c91001d7e1 chore: refresh uv.lock file to fix CI/CD build issues
As requested by maintainer to resolve integration test failures
2025-07-28 22:56:07 -04:00
Neovasky 83748a27da fix: apply ruff formatting to pass CI checks
- Fixed line length issues in logger.warning calls
- Removed trailing spaces in docstrings
- Applied consistent formatting across all files
2025-07-28 11:52:10 -04:00
Neovasky 3ddeeab67f fix(calendar): address PR feedback from maintainer
- Remove CHANGELOG.md changes (auto-generated from commits)
- Move all parameter descriptions into function docstrings for LLM context
- Remove unused caldav dependency (using httpx for CalDAV implementation)
- Move datetime imports to top of modules
- Remove load_dotenv from tests/conftest.py
- Clarify Event vs Meeting distinction in docstrings
- Handle 401 auth errors gracefully in calendar tests

Addresses all feedback from PR #95 review
2025-07-28 11:44:53 -04:00
Chris Coutinho a2c78ee1ef test: Add tests for MCP tools and resources 2025-07-27 17:43:55 +02:00
Chris Coutinho 1e19061ee0 chore: Move tools into separate modules 2025-07-27 14:11:02 +02:00
Neovasky 2e078498b1 refactor(calendar): optimize logging for production readiness
- Change routine operation logs from info to debug level
- Simplify success messages for better readability
- Remove redundant calendar/path information from log messages
- Align logging style with repository standards

Following patterns established by repository maintainer in WebDAV client cleanup.
2025-07-27 00:46:57 -04:00
Neovasky 7291c930c4 feat(calendar): add comprehensive Calendar app support via CalDAV protocol
- Add complete CalDAV client implementation following NextCloud patterns
- Implement 11 comprehensive calendar MCP tools:
  * nc_calendar_list_calendars - list available calendars
  * nc_calendar_create_event - full event creation with recurrence, reminders, attendees
  * nc_calendar_list_events - enhanced with advanced filtering capabilities
  * nc_calendar_get_event - detailed event information retrieval
  * nc_calendar_update_event - comprehensive event modification
  * nc_calendar_delete_event - event removal
  * nc_calendar_create_meeting - quick meeting creation with smart defaults
  * nc_calendar_get_upcoming_events - upcoming events in next N days
  * nc_calendar_find_availability - intelligent scheduling with conflict detection
  * nc_calendar_bulk_operations - batch update/delete/move operations
  * nc_calendar_manage_calendar - calendar creation and management

- Add CalDAV and iCalendar dependencies to support calendar operations
- Implement comprehensive integration tests (11 test cases covering all scenarios)
- Update documentation with complete calendar tools reference and usage examples

Resolves #74
2025-07-27 00:25:31 -04:00
github-actions[bot] b8191c134a bump: version 0.4.1 → 0.5.0 2025-07-26 11:32:13 +00:00
Chris Coutinho 09061d9e4f Merge pull request #94 from cbcoutinho/fix/webdav
Update webdav client create_directory method to handle recursiv…
2025-07-26 13:31:50 +02:00
Chris Coutinho 2d3cb85fb2 Merge pull request #92 from neovasky/master
feat(webdav): add complete file system support
2025-07-26 13:28:12 +02:00
Chris Coutinho 3ad07d05dd feat: Update webdav client create_directory method to handle recursive directories 2025-07-26 13:27:21 +02:00
Neovasky 50c1215676 fix: apply ruff formatting to test_webdav_operations.py
- Fix quote style from single to double quotes
- Improve line breaks and spacing for better readability
- Address CI formatting requirements

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-07-26 02:33:21 -04:00
Neovasky bf5879d408 test: add comprehensive WebDAV integration tests
- Add 8 core WebDAV operation tests covering CRUD operations
- Add complex attachment cleanup test for category changes
- Fix ruff formatting violations in webdav.py and server.py
- Address PR feedback requirements for expanded WebDAV functionality

Tests focus on WebDAV client functionality and run locally with docker-compose.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-07-26 02:28:13 -04:00
Chris Coutinho 442e82e994 Merge pull request #88 from cbcoutinho/renovate/redis-alpine
chore(deps): update redis:alpine docker digest to 25c0ae3
2025-07-25 11:05:54 +02:00
Neovasky 9e96999f02 feat(webdav): add complete file system support
- Add nc_webdav_list_directory tool for browsing any NextCloud directory
  - Add nc_webdav_read_file tool with automatic text/binary content handling
  - Add nc_webdav_write_file tool supporting text and base64 binary content
  - Add nc_webdav_create_directory tool for creating directories
  - Add nc_webdav_delete_resource tool for deleting files and directories
  - Extend WebDAV client beyond Notes attachments to general file operations
  - Add XML parsing for WebDAV PROPFIND responses with metadata extraction
  - Improve type annotations throughout codebase for better IDE support
  - Add comprehensive documentation with usage examples

  This transforms the NextCloud MCP server from a limited Notes/Tables tool
  into a full-featured file system interface, enabling complete NextCloud
  file management through LLM interactions.
2025-07-25 03:15:52 -04:00
Chris Coutinho e983693534 Merge pull request #90 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.3
2025-07-25 01:57:59 +02:00
renovate-bot-cbcoutinho[bot] b8a14a2229 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.3 2025-07-24 22:13:40 +00:00
Chris Coutinho 508f83dfad Merge pull request #89 from cbcoutinho/renovate/nextcloud-31.0.7
chore(deps): update nextcloud:31.0.7 docker digest to 31d564f
2025-07-24 14:22:55 +02:00
renovate-bot-cbcoutinho[bot] ce8d5f92b1 chore(deps): update nextcloud:31.0.7 docker digest to 31d564f 2025-07-24 04:11:59 +00:00
Chris Coutinho ca32ff39b8 Merge pull request #91 from cbcoutinho/renovate/astral-sh-setup-uv-digest
chore(deps): update astral-sh/setup-uv digest to e92bafb
2025-07-24 01:38:53 +02:00
renovate-bot-cbcoutinho[bot] 9da53e51f0 chore(deps): update astral-sh/setup-uv digest to e92bafb 2025-07-23 22:14:26 +00:00
Chris Coutinho 2cbac7c4be Merge pull request #82 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.0
2025-07-18 23:28:51 +02:00
Chris Coutinho d2394465d7 Merge pull request #87 from cbcoutinho/renovate/astral-sh-setup-uv-digest
chore(deps): update astral-sh/setup-uv digest to 7edac99
2025-07-18 23:27:37 +02:00
renovate-bot-cbcoutinho[bot] c2615ac24d chore(deps): update astral-sh/setup-uv digest to 7edac99 2025-07-18 10:12:13 +00:00
renovate-bot-cbcoutinho[bot] 62e21f1f94 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.0 2025-07-18 04:14:53 +00:00
renovate-bot-cbcoutinho[bot] 9bd95a8b17 chore(deps): update redis:alpine docker digest to 25c0ae3 2025-07-17 22:08:58 +00:00
Chris Coutinho bfd2eed97b Merge pull request #85 from cbcoutinho/renovate/mariadb-lts
chore(deps): update mariadb:lts docker digest to 2bcbaec
2025-07-16 23:21:42 +02:00
renovate-bot-cbcoutinho[bot] 8a0b964add chore(deps): update mariadb:lts docker digest to 2bcbaec 2025-07-16 16:05:48 +00:00
Chris Coutinho 59bab51090 Merge pull request #83 from cbcoutinho/renovate/mariadb-lts
chore(deps): update mariadb:lts docker digest to ee8fadc
2025-07-16 08:39:04 +02:00
Chris Coutinho 12fa550b60 Merge pull request #84 from cbcoutinho/renovate/redis-alpine
chore(deps): update redis:alpine docker digest to d12963a
2025-07-16 08:37:44 +02:00
renovate-bot-cbcoutinho[bot] 85cdf75a5b chore(deps): update redis:alpine docker digest to d12963a 2025-07-16 04:08:55 +00:00
renovate-bot-cbcoutinho[bot] 0ee2b5b034 chore(deps): update mariadb:lts docker digest to ee8fadc 2025-07-16 04:08:51 +00:00
Chris Coutinho 0c4d140bb9 Merge pull request #81 from cbcoutinho/renovate/nextcloud-31.x
chore(deps): update nextcloud docker tag to v31.0.7
2025-07-13 23:13:09 +02:00
renovate-bot-cbcoutinho[bot] f515d74a4d chore(deps): update nextcloud docker tag to v31.0.7 2025-07-12 04:05:32 +00:00
github-actions[bot] 79835b3439 bump: version 0.4.0 → 0.4.1 2025-07-10 17:35:22 +00:00
Chris Coutinho d518b76878 Merge pull request #64 from cbcoutinho/renovate/mcp-1.x
fix(deps): update dependency mcp to >=1.10,<1.11
2025-07-10 19:34:58 +02:00
Chris Coutinho 5179db40db Merge pull request #79 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.20
2025-07-10 07:27:03 +02:00
renovate-bot-cbcoutinho[bot] 9cbeecae64 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.20 2025-07-09 22:06:38 +00:00
Chris Coutinho c5af81c94f Merge pull request #78 from cbcoutinho/cbcoutinho-patch-2
chore: Update README.md
2025-07-09 09:55:16 +02:00
Chris Coutinho ae966710a9 Update README.md 2025-07-09 09:54:58 +02:00
Chris Coutinho 9b14135dd3 Update README.md 2025-07-09 09:54:24 +02:00
Chris Coutinho 6f92cd8157 chore: Update README.md 2025-07-09 09:53:45 +02:00
Chris Coutinho 6545f8165f (chore) Update README.md 2025-07-09 00:36:02 +02:00
Chris Coutinho 4a742442fb Merge pull request #77 from cbcoutinho/renovate/redis-alpine
chore(deps): update redis:alpine docker digest to 73734b0
2025-07-08 09:18:28 +02:00
renovate-bot-cbcoutinho[bot] f84144fcaa chore(deps): update redis:alpine docker digest to 73734b0 2025-07-07 22:04:16 +00:00
Chris Coutinho e09f373f84 Merge pull request #76 from cbcoutinho/refactor/clients
Move clients into separate submodule
2025-07-07 00:09:39 +02:00
Chris Coutinho e50be7db07 chore: Move clients into separate submodule 2025-07-07 00:06:24 +02:00
Chris Coutinho f03ab4ef55 chore: [skip ci] Remove tables-openapi.json 2025-07-06 09:53:33 +02:00
Chris Coutinho 3d26c6c145 Merge pull request #68 from cbcoutinho/renovate/mariadb-lts
chore(deps): update mariadb:lts docker digest to 1e4ec03
2025-07-06 09:51:22 +02:00
Chris Coutinho a4b0c84f79 Merge pull request #67 from cbcoutinho/renovate/nextcloud-31.0.6
chore(deps): update nextcloud:31.0.6 docker digest to 588609d
2025-07-06 09:51:13 +02:00
Chris Coutinho e67e7c4246 Merge pull request #69 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.19
2025-07-06 09:51:05 +02:00
Chris Coutinho e0c4cc5d77 Merge pull request #70 from cbcoutinho/renovate/hoverkraft-tech-compose-action-2.x
chore(deps): update hoverkraft-tech/compose-action action to v2.3.0
2025-07-06 09:50:57 +02:00
github-actions[bot] b43ffad708 bump: version 0.3.0 → 0.4.0 2025-07-06 07:50:10 +00:00
Chris Coutinho cab7a59d2b Merge pull request #71 from cbcoutinho/feature/tables-app
Initialize Tables App
2025-07-06 09:49:45 +02:00
Chris Coutinho ca5bbb783a fix: update tests 2025-07-06 09:40:27 +02:00
Chris Coutinho d47e2bb8f0 test: Update tests with updated API 2025-07-06 09:37:31 +02:00
Chris Coutinho a1c186aa95 feat: Add TablesClient and associated tools 2025-07-06 09:18:34 +02:00
Chris Coutinho 57440f845f chore: Update pre-commit 2025-07-06 08:42:09 +02:00
Chris Coutinho a57c12591a chore: ruff format 2025-07-06 08:41:02 +02:00
Chris Coutinho 5b512f83bd refactor: Modularize NC and Notes app client 2025-07-06 08:39:28 +02:00
renovate-bot-cbcoutinho[bot] 4a2fd67e51 chore(deps): update hoverkraft-tech/compose-action action to v2.3.0 2025-07-05 13:12:44 +00:00
renovate-bot-cbcoutinho[bot] da3a0049a0 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.19 2025-07-05 13:12:37 +00:00
renovate-bot-cbcoutinho[bot] bb53ba6275 chore(deps): update nextcloud:31.0.6 docker digest to 588609d 2025-07-05 13:12:33 +00:00
renovate-bot-cbcoutinho[bot] 7a6c7c6efa chore(deps): update mariadb:lts docker digest to 1e4ec03 2025-07-05 13:12:28 +00:00
Chris Coutinho 266d2dac8d Merge pull request #66 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.17
2025-06-30 08:32:41 +02:00
renovate-bot-cbcoutinho[bot] d64c6e112e chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.17 2025-06-29 16:04:31 +00:00
Chris Coutinho 167517b95d Merge pull request #65 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.16
2025-06-29 00:20:35 +02:00
renovate-bot-cbcoutinho[bot] 33aa778713 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.16 2025-06-27 22:06:33 +00:00
renovate-bot-cbcoutinho[bot] 251c9aaae6 fix(deps): update dependency mcp to >=1.10,<1.11 2025-06-26 16:06:09 +00:00
Chris Coutinho ded48acd31 Merge pull request #63 from cbcoutinho/renovate/nextcloud-31.0.6
chore(deps): update nextcloud:31.0.6 docker digest to 0b133af
2025-06-26 14:01:55 +02:00
renovate-bot-cbcoutinho[bot] 0dacd84cc2 chore(deps): update nextcloud:31.0.6 docker digest to 0b133af 2025-06-26 10:07:22 +00:00
Chris Coutinho c0782dc69e Merge pull request #61 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.15
2025-06-26 09:44:21 +02:00
Chris Coutinho 4a8f9f7f7e chore: Update with "mergeConfidence:all-badges" 2025-06-26 09:43:59 +02:00
Chris Coutinho db9f2cad43 Merge pull request #62 from cbcoutinho/renovate/nextcloud-31.0.6
chore(deps): update nextcloud:31.0.6 docker digest to dff5690
2025-06-26 08:15:54 +02:00
renovate-bot-cbcoutinho[bot] d52860c86d chore(deps): update nextcloud:31.0.6 docker digest to dff5690 2025-06-26 04:06:33 +00:00
renovate-bot-cbcoutinho[bot] 4992f700c6 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.15 2025-06-25 16:06:41 +00:00
Chris Coutinho cc2777210b Merge pull request #60 from cbcoutinho/renovate/astral-sh-setup-uv-digest
chore(deps): update astral-sh/setup-uv digest to bd01e18
2025-06-25 14:40:44 +02:00
renovate-bot-cbcoutinho[bot] ad1320319b chore(deps): update astral-sh/setup-uv digest to bd01e18 2025-06-25 10:08:27 +00:00
Chris Coutinho 9d9f1e1eaa Merge pull request #53 from cbcoutinho/renovate/nextcloud-31.x
chore(deps): update nextcloud docker tag to v31.0.6
2025-06-24 14:44:13 +02:00
Chris Coutinho 7b3b624403 Merge pull request #59 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.14
2025-06-24 14:44:04 +02:00
renovate-bot-cbcoutinho[bot] 5c908bf8d2 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.14 2025-06-24 12:28:24 +00:00
Chris Coutinho fe16f4db54 Merge pull request #58 from cbcoutinho/renovate/astral-sh-setup-uv-digest
chore(deps): update astral-sh/setup-uv digest to 445689e
2025-06-20 07:26:00 +02:00
renovate-bot-cbcoutinho[bot] 7b10296058 chore(deps): update astral-sh/setup-uv digest to 445689e 2025-06-19 22:09:17 +00:00
Chris Coutinho e6890ab24d Merge pull request #57 from cbcoutinho/renovate/astral-sh-setup-uv-digest
chore(deps): update astral-sh/setup-uv digest to a02a550
2025-06-19 13:40:18 +02:00
renovate-bot-cbcoutinho[bot] cf49866a87 chore(deps): update astral-sh/setup-uv digest to a02a550 2025-06-18 22:12:49 +00:00
Chris Coutinho d8e7d0b465 Merge pull request #55 from cbcoutinho/renovate/docker-setup-buildx-action-digest
chore(deps): update docker/setup-buildx-action digest to e468171
2025-06-18 22:27:27 +02:00
renovate-bot-cbcoutinho[bot] c336c5d2a2 chore(deps): update docker/setup-buildx-action digest to e468171 2025-06-18 10:11:18 +00:00
Chris Coutinho 45c0622459 Merge pull request #56 from lwsinclair/add-mseep-badge
Add MseeP.ai badge
2025-06-17 15:16:34 +02:00
Lawrence Sinclair 7dfbe9dd62 Add MseeP.ai badge to README.md 2025-06-17 12:15:29 +07:00
renovate-bot-cbcoutinho[bot] 3d5da56d83 chore(deps): update nextcloud docker tag to v31.0.6 2025-06-14 04:11:21 +00:00
Chris Coutinho 2b1dbfef39 Merge pull request #51 from cbcoutinho/renovate/nextcloud-31.0.5
chore(deps): update nextcloud:31.0.5 docker digest to 3aed4aa
2025-06-13 11:58:37 +02:00
Chris Coutinho 2e016080fd Merge pull request #52 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.13
2025-06-13 11:58:27 +02:00
renovate-bot-cbcoutinho[bot] e0a966b4a6 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.13 2025-06-12 22:12:42 +00:00
renovate-bot-cbcoutinho[bot] 07a8b6e704 chore(deps): update nextcloud:31.0.5 docker digest to 3aed4aa 2025-06-12 22:12:38 +00:00
Chris Coutinho 659da9a770 Merge pull request #50 from cbcoutinho/renovate/nextcloud-31.0.5
chore(deps): update nextcloud:31.0.5 docker digest to 21780a1
2025-06-11 18:12:11 +02:00
renovate-bot-cbcoutinho[bot] 18f8b73982 chore(deps): update nextcloud:31.0.5 docker digest to 21780a1 2025-06-11 16:07:25 +00:00
Chris Coutinho 2bc0988e8d Merge pull request #48 from cbcoutinho/renovate/mariadb-lts
chore(deps): update mariadb:lts docker digest to 1e66902
2025-06-11 08:51:27 +02:00
Chris Coutinho 74235ed8bb Merge pull request #49 from cbcoutinho/renovate/nextcloud-31.0.5
chore(deps): update nextcloud:31.0.5 docker digest to f43cee6
2025-06-11 08:51:20 +02:00
Chris Coutinho 89a9af7c25 Merge pull request #47 from cbcoutinho/renovate/softprops-action-gh-release-digest
chore(deps): update softprops/action-gh-release digest to 72f2c25
2025-06-11 08:51:10 +02:00
renovate-bot-cbcoutinho[bot] d247a07643 chore(deps): update softprops/action-gh-release digest to 72f2c25 2025-06-11 04:07:40 +00:00
renovate-bot-cbcoutinho[bot] 794d4184d2 chore(deps): update nextcloud:31.0.5 docker digest to f43cee6 2025-06-11 04:07:35 +00:00
renovate-bot-cbcoutinho[bot] cc17b28eab chore(deps): update mariadb:lts docker digest to 1e66902 2025-06-11 04:07:30 +00:00
Chris Coutinho 5626f6fd6f Merge pull request #46 from cbcoutinho/renovate/softprops-action-gh-release-digest
chore(deps): update softprops/action-gh-release digest to d5382d3
2025-06-10 09:10:37 +02:00
renovate-bot-cbcoutinho[bot] 79a466d16c chore(deps): update softprops/action-gh-release digest to d5382d3 2025-06-10 04:07:22 +00:00
Chris Coutinho 6aa06b4c9d Merge pull request #45 from cbcoutinho/renovate/ghcr.io-astral-sh-uv-0.x
chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.12
2025-06-07 07:35:17 +02:00
Chris Coutinho c993872ab5 Merge pull request #44 from cbcoutinho/renovate/nextcloud-31.0.5
chore(deps): update nextcloud:31.0.5 docker digest to e775d46
2025-06-07 07:35:08 +02:00
renovate-bot-cbcoutinho[bot] e69819a49b chore(deps): update nextcloud:31.0.5 docker digest to e775d46 2025-06-07 04:06:37 +00:00
renovate-bot-cbcoutinho[bot] 49868d2bb5 chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.7.12 2025-06-06 22:05:10 +00:00
github-actions[bot] 33c8623d5c bump: version 0.2.5 → 0.3.0 2025-06-06 17:20:50 +00:00
Chris Coutinho 150e656a36 Merge pull request #43 from cbcoutinho/feature/async
Switch to using async client
2025-06-06 19:20:25 +02:00
45 changed files with 6200 additions and 942 deletions
+1 -1
View File
@@ -25,7 +25,7 @@ jobs:
github_token: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
changelog_increment_filename: body.md
- name: Release
uses: softprops/action-gh-release@da05d552573ad5aba039eaac05058a918a7bf631 # v2
uses: softprops/action-gh-release@72f2c25fcb47643c292f7107632f7a47c1df5cd8 # v2
with:
body_path: "body.md"
tag_name: v${{ env.REVISION }}
+2 -2
View File
@@ -16,7 +16,7 @@ jobs:
- name: Docker meta
id: meta
uses: docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804 # v5
uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5
with:
# list of Docker images to use as base name for tags
images: |
@@ -33,7 +33,7 @@ jobs:
type=raw,value=latest,enable={{is_default_branch}}
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@b5ca514318bd6ebac0fb2aedd5d36ec1b5c232a2 # v3
uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3
- name: Log in to GitHub Container Registry
if: github.event_name != 'pull_request'
+3 -3
View File
@@ -11,7 +11,7 @@ jobs:
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
- name: Install the latest version of uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v6
uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6
- name: Check format
run: |
uv run --frozen ruff format --diff
@@ -27,11 +27,11 @@ jobs:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
- name: Run docker compose
uses: hoverkraft-tech/compose-action@8be2d741e891ac9b8ac20825e6f3904149599925 # v2.2.0
uses: hoverkraft-tech/compose-action@40041ff1b97dbf152cd2361138c2b03fa29139df # v2.3.0
with:
compose-file: "./docker-compose.yml"
- name: Install the latest version of uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v6
uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6
- name: Wait for service to be ready
run: |
+4
View File
@@ -1,2 +1,6 @@
__pycache__/
.coverage
.env
*.env
.env.local
.env.*.local
+8 -3
View File
@@ -1,8 +1,13 @@
repos:
- hooks:
- repo: https://github.com/commitizen-tools/commitizen
rev: v4.8.3
hooks:
- id: commitizen
- id: commitizen-branch
stages:
- pre-push
repo: https://github.com/commitizen-tools/commitizen
rev: v4.8.2
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.12.5
hooks:
- id: ruff-check
- id: ruff-format
+59
View File
@@ -1,3 +1,62 @@
## v0.6.1 (2025-08-01)
### Fix
- **calendar**: Fix iCalendar date vs datetime format
- **calendar**: Remove try/except in calendar API
## v0.6.0 (2025-07-29)
### Feat
- **calendar**: add comprehensive Calendar app support via CalDAV protocol
### Fix
- apply ruff formatting to pass CI checks
- **calendar**: address PR feedback from maintainer
### Refactor
- **calendar**: optimize logging for production readiness
## v0.5.0 (2025-07-26)
### Feat
- Update webdav client create_directory method to handle recursive directories
- **webdav**: add complete file system support
### Fix
- apply ruff formatting to test_webdav_operations.py
## v0.4.1 (2025-07-10)
### Fix
- **deps**: update dependency mcp to >=1.10,<1.11
## v0.4.0 (2025-07-06)
### Feat
- Add TablesClient and associated tools
### Fix
- update tests
### Refactor
- Modularize NC and Notes app client
## v0.3.0 (2025-06-06)
### Feat
- Switch to using async client
## v0.2.5 (2025-05-25)
### Fix
+2 -2
View File
@@ -1,4 +1,4 @@
FROM ghcr.io/astral-sh/uv:0.7.11-python3.11-alpine@sha256:66d4d13288afecfeb2173b267a6c0765957d2122935c447d6963ea7b38929a99
FROM ghcr.io/astral-sh/uv:0.8.4-python3.11-alpine@sha256:f2c5b953b713f455bcac4429303bb21d7d2547d56a64e1a7b2517cc9f0563f0f
WORKDIR /app
@@ -6,4 +6,4 @@ COPY . .
RUN uv sync --locked --no-dev
CMD ["/app/.venv/bin/mcp", "run", "--transport", "sse", "/app/nextcloud_mcp_server/server.py:mcp"]
CMD ["/app/.venv/bin/mcp", "run", "--transport", "sse", "/app/nextcloud_mcp_server/app.py:mcp"]
+207 -17
View File
@@ -6,24 +6,209 @@ The Nextcloud MCP (Model Context Protocol) server allows Large Language Models (
## Features
Currently, the server primarily interacts with the Nextcloud Notes API, providing tools and resources to manage notes.
The server provides integration with multiple Nextcloud apps, enabling LLMs to interact with your Nextcloud data through a rich set of tools and resources.
### Available Tools
## Supported Nextcloud Apps
* `nc_notes_create_note`: Create a new note.
* `nc_notes_update_note`: Update an existing note by ID.
* `nc_notes_append_content`: Append content to an existing note with a clear separator.
* `nc_notes_delete_note`: Delete a note by ID.
* `nc_notes_search_notes`: Search notes by title or content.
* `nc_get_note`: Get a specific note by ID.
| App | Support Status | Description |
|-----|----------------|-------------|
| **Notes** | ✅ Full Support | Create, read, update, delete, and search notes. Handle attachments via WebDAV. |
| **Calendar** | ✅ Full Support | Complete calendar integration - create, update, delete events. Support for recurring events, reminders, attendees, and all-day events via CalDAV. |
| **Tables** | ⚠️ Row Operations | Read table schemas and perform CRUD operations on table rows. Table management not yet supported. |
| **Files (WebDAV)** | ✅ Full Support | Complete file system access - browse directories, read/write files, create/delete resources. |
| **Contacts** | ✅ Full Support | Create, read, update, and delete contacts and address books via CardDAV. |
### Available Resources
## Available Tools
* `notes://{note_id}`: Access a specific note by its ID.
* `notes://all`: Access all notes.
* `notes://settings`: Access note settings.
* `nc://capabilities`: Access Nextcloud server capabilities.
* `nc://Notes/{note_id}/attachments/{attachment_filename}`: Access attachments for notes.
### Notes Tools
| Tool | Description |
|------|-------------|
| `nc_get_note` | Get a specific note by ID |
| `nc_notes_create_note` | Create a new note with title, content, and category |
| `nc_notes_update_note` | Update an existing note by ID |
| `nc_notes_append_content` | Append content to an existing note with a clear separator |
| `nc_notes_delete_note` | Delete a note by ID |
| `nc_notes_search_notes` | Search notes by title or content |
### Calendar Tools
| Tool | Description |
|------|-------------|
| `nc_calendar_list_calendars` | List all available calendars for the user |
| `nc_calendar_create_event` | Create a comprehensive calendar event with full feature support (recurring, reminders, attendees, etc.) |
| `nc_calendar_list_events` | **Enhanced:** List events with advanced filtering (min attendees, duration, categories, status, search across all calendars) |
| `nc_calendar_get_event` | Get detailed information about a specific event |
| `nc_calendar_update_event` | Update any aspect of an existing event |
| `nc_calendar_delete_event` | Delete a calendar event |
| `nc_calendar_create_meeting` | Quick meeting creation with smart defaults |
| `nc_calendar_get_upcoming_events` | Get upcoming events in the next N days |
| `nc_calendar_find_availability` | **New:** Intelligent availability finder - find free time slots for meetings with attendee conflict detection |
| `nc_calendar_bulk_operations` | **New:** Bulk update, delete, or move events matching filter criteria |
| `nc_calendar_manage_calendar` | **New:** Create, delete, and manage calendar properties |
### Contacts Tools
| Tool | Description |
|------|-------------|
| `nc_contacts_list_addressbooks` | List all available addressbooks for the user |
| `nc_contacts_list_contacts` | List all contacts in a specific addressbook |
| `nc_contacts_create_addressbook` | Create a new addressbook |
| `nc_contacts_delete_addressbook` | Delete an addressbook |
| `nc_contacts_create_contact` | Create a new contact in an addressbook |
| `nc_contacts_delete_contact` | Delete a contact from an addressbook |
### Tables Tools
| Tool | Description |
|------|-------------|
| `nc_tables_list_tables` | List all tables available to the user |
| `nc_tables_get_schema` | Get the schema/structure of a specific table including columns and views |
| `nc_tables_read_table` | Read rows from a table with optional pagination |
| `nc_tables_insert_row` | Insert a new row into a table |
| `nc_tables_update_row` | Update an existing row in a table |
| `nc_tables_delete_row` | Delete a row from a table |
### WebDAV File System Tools
| Tool | Description |
|------|-------------|
| `nc_webdav_list_directory` | List files and directories in any NextCloud path |
| `nc_webdav_read_file` | Read file content (text files decoded, binary as base64) |
| `nc_webdav_write_file` | Create or update files in NextCloud |
| `nc_webdav_create_directory` | Create new directories |
| `nc_webdav_delete_resource` | Delete files or directories |
## Available Resources
| Resource | Description |
|----------|-------------|
| `nc://capabilities` | Access Nextcloud server capabilities |
| `notes://settings` | Access Notes app settings |
| `nc://Notes/{note_id}/attachments/{attachment_filename}` | Access attachments for notes |
### WebDAV File System Access
The server provides complete file system access to your NextCloud instance, enabling you to:
- Browse any directory structure
- Read and write files of any type
- Create and delete directories
- Manage your NextCloud files directly through LLM interactions
**Usage Examples:**
```python
# List files in root directory
await nc_webdav_list_directory("")
# Browse a specific folder
await nc_webdav_list_directory("Documents/Projects")
# Read a text file
content = await nc_webdav_read_file("Documents/readme.txt")
# Create a new directory
await nc_webdav_create_directory("NewProject/docs")
# Write content to a file
await nc_webdav_write_file("NewProject/docs/notes.md", "# My Notes\n\nContent here...")
# Delete a file or directory
await nc_webdav_delete_resource("old_file.txt")
```
### Calendar Integration
The server provides comprehensive calendar integration through CalDAV, enabling you to:
- List all available calendars
- Create, read, update, and delete calendar events
- Handle recurring events with RRULE support
- Manage event reminders and notifications
- Support all-day and timed events
- Handle attendees and meeting invitations
- Organize events with categories and priorities
**Usage Examples:**
```python
# List available calendars
calendars = await nc_calendar_list_calendars()
# Create a simple event
await nc_calendar_create_event(
calendar_name="personal",
title="Team Meeting",
start_datetime="2025-07-28T14:00:00",
end_datetime="2025-07-28T15:00:00",
description="Weekly team sync",
location="Conference Room A"
)
# Create a recurring weekly meeting
await nc_calendar_create_event(
calendar_name="work",
title="Weekly Standup",
start_datetime="2025-07-28T09:00:00",
end_datetime="2025-07-28T09:30:00",
recurring=True,
recurrence_rule="FREQ=WEEKLY;BYDAY=MO"
)
# Quick meeting creation
await nc_calendar_create_meeting(
title="Client Call",
date="2025-07-28",
time="15:00",
duration_minutes=60,
attendees="client@example.com,colleague@company.com"
)
# Get upcoming events
events = await nc_calendar_get_upcoming_events(days_ahead=7)
# Advanced search - find all meetings with 5+ attendees lasting 2+ hours
long_meetings = await nc_calendar_list_events(
calendar_name="", # Search all calendars
search_all_calendars=True,
start_date="2025-07-01",
end_date="2025-07-31",
min_attendees=5,
min_duration_minutes=120,
title_contains="meeting"
)
# Find availability for a 1-hour meeting with specific attendees
availability = await nc_calendar_find_availability(
duration_minutes=60,
attendees="sarah@company.com,mike@company.com",
date_range_start="2025-07-28",
date_range_end="2025-08-04",
business_hours_only=True,
exclude_weekends=True,
preferred_times="09:00-12:00,14:00-17:00"
)
# Bulk update all team meetings to new location
bulk_result = await nc_calendar_bulk_operations(
operation="update",
title_contains="team meeting",
start_date="2025-08-01",
end_date="2025-08-31",
new_location="Conference Room B",
new_reminder_minutes=15
)
# Create a new project calendar
new_calendar = await nc_calendar_manage_calendar(
action="create",
calendar_name="project-alpha",
display_name="Project Alpha Calendar",
description="Calendar for Project Alpha team",
color="#FF5722"
)
```
### Note Attachments
@@ -70,6 +255,7 @@ NEXTCLOUD_PASSWORD=your_nextcloud_app_password_or_login_password
* `NEXTCLOUD_HOST`: The full URL of your Nextcloud instance.
* `NEXTCLOUD_USERNAME`: Your Nextcloud username.
* `NEXTCLOUD_PASSWORD`: **Important:** It is highly recommended to use a dedicated Nextcloud App Password for security. You can generate one in your Nextcloud Security settings. Alternatively, you can use your regular login password, but this is less secure.
* `FASTMCP_HOST`: _Optional:_ By default FastMCP binds to localhost. Use this variable to set a different binding address (e.g. `0.0.0.0`)
## Running the Server
@@ -82,10 +268,12 @@ Ensure your environment variables are loaded, then run the server using `mcp run
export $(grep -v '^#' .env | xargs)
# Run the server
mcp run --transport sse nextcloud_mcp_server.server:mcp
mcp run --transport sse nextcloud_mcp_server.app:mcp
```
The server will start, typically listening on `http://0.0.0.0:8000`.
The server will start, typically listening on `http://localhost:8000`.
> NOTE: To make the server bind to a different address, use the FASTMCP_HOST environmental variable
### Using Docker
@@ -117,4 +305,6 @@ Contributions are welcome! Please feel free to submit issues or pull requests on
## License
This project is licensed under the MIT License. See the LICENSE file for details.
This project is licensed under the AGPL-3.0 License. See the [LICENSE](./LICENSE) file for details.
[![MseeP.ai Security Assessment Badge](https://mseep.net/pr/cbcoutinho-nextcloud-mcp-server-badge.png)](https://mseep.ai/app/cbcoutinho-nextcloud-mcp-server)
+29
View File
@@ -0,0 +1,29 @@
#!/bin/bash
set -e # Exit on any error
echo "Installing and configuring Calendar app..."
# Enable calendar app
php /var/www/html/occ app:enable calendar
# Wait for calendar app to be fully initialized
echo "Waiting for calendar app to initialize..."
sleep 5
# Ensure maintenance mode is off before calendar operations
php /var/www/html/occ maintenance:mode --off
# Sync DAV system to ensure proper initialization
echo "Syncing DAV system..."
php /var/www/html/occ dav:sync-system-addressbook
# Repair calendar app to ensure proper setup
echo "Repairing calendar app..."
php /var/www/html/occ maintenance:repair --include-expensive
# Final wait to ensure CalDAV service is fully ready
echo "Final CalDAV initialization wait..."
sleep 5
echo "Calendar app installation complete!"
+3
View File
@@ -0,0 +1,3 @@
#!/bin/bash
php /var/www/html/occ app:enable contacts
+3
View File
@@ -0,0 +1,3 @@
#!/bin/bash
php /var/www/html/occ app:enable tables
+4 -3
View File
@@ -3,7 +3,7 @@ services:
# https://hub.docker.com/_/mariadb
db:
# Note: Check the recommend version here: https://docs.nextcloud.com/server/latest/admin_manual/installation/system_requirements.html#server
image: mariadb:lts@sha256:1d18f91deb21136d1881705720071d1b474a9904ecca827058bf1c0fc64d3118
image: mariadb:lts@sha256:2bcbaec92bd9d4f6591bc8103d3a8e6d0512ee2235506e47a2e129d190444405
restart: always
command: --transaction-isolation=READ-COMMITTED
volumes:
@@ -17,11 +17,11 @@ services:
# Note: Redis is an external service. You can find more information about the configuration here:
# https://hub.docker.com/_/redis
redis:
image: redis:alpine@sha256:48501c5ad00d5563bc30c075c7bcef41d7d98de3e9a1e6c752068c66f0a8463b
image: redis:alpine@sha256:25c0ae32c6c2301798579f5944af53729766a18eff5660bbef196fc2e6214a9c
restart: always
app:
image: nextcloud:31.0.5@sha256:3f71577339ef1db0d1900c8574853d11fa7100452bf24f0a06fae5d9ee019cb4
image: nextcloud:31.0.7@sha256:81dc361f8f216d8acff20bd3dea2226fb6cea883c277505cbb2ddd6327c867fa
#user: www-data:www-data
restart: always
#post_start:
@@ -52,6 +52,7 @@ services:
- NEXTCLOUD_HOST=http://app:80
- NEXTCLOUD_USERNAME=admin
- NEXTCLOUD_PASSWORD=admin
- FASTMCP_HOST=0.0.0.0
volumes:
nextcloud:
+64
View File
@@ -0,0 +1,64 @@
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import setup_logging
from nextcloud_mcp_server.server import (
configure_calendar_tools,
configure_contacts_tools,
configure_notes_tools,
configure_tables_tools,
configure_webdav_tools,
)
setup_logging()
@dataclass
class AppContext:
client: NextcloudClient
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
logging.info("Creating Nextcloud client")
client = NextcloudClient.from_env()
logging.info("Client initialization wait complete.")
try:
yield AppContext(client=client)
finally:
# Cleanup on shutdown
await client.close()
# Create an MCP server
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan)
logger = logging.getLogger(__name__)
@mcp.resource("nc://capabilities")
async def nc_get_capabilities():
"""Get the Nextcloud Host capabilities"""
ctx: Context = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.capabilities()
configure_notes_tools(mcp)
configure_tables_tools(mcp)
configure_webdav_tools(mcp)
configure_calendar_tools(mcp)
configure_contacts_tools(mcp)
def run():
mcp.run()
-674
View File
@@ -1,674 +0,0 @@
import os
import mimetypes
from httpx import (
AsyncClient,
Auth,
BasicAuth,
Request,
Response,
HTTPStatusError,
)
import logging
logger = logging.getLogger(__name__)
def log_request(request: Request):
logger.info(
"Request event hook: %s %s - Waiting for content",
request.method,
request.url,
)
logger.info("Request body: %s", request.content)
logger.info("Headers: %s", request.headers)
def log_response(response: Response):
response.read() # Explicitly read the stream before accessing .text
logger.info("Response [%s] %s", response.status_code, response.text)
class NextcloudClient:
def __init__(self, base_url: str, username: str, auth: Auth | None = None):
self.username = username # Store username
self._client = AsyncClient(
base_url=base_url,
auth=auth,
# event_hooks={"request": [log_request], "response": [log_response]},
)
@classmethod
def from_env(cls):
logger.info("Creating NC Client using env vars")
host = os.environ["NEXTCLOUD_HOST"]
username = os.environ["NEXTCLOUD_USERNAME"]
password = os.environ["NEXTCLOUD_PASSWORD"]
# Pass username to constructor
return cls(base_url=host, username=username, auth=BasicAuth(username, password))
async def capabilities(self):
response = await self._client.get(
"/ocs/v2.php/cloud/capabilities",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
)
response.raise_for_status()
return response.json()
async def notes_get_settings(self):
response = await self._client.get("/apps/notes/api/v1/settings")
response.raise_for_status()
return response.json()
async def notes_get_all(self):
response = await self._client.get("/apps/notes/api/v1/notes")
response.raise_for_status()
return response.json()
async def notes_get_note(self, *, note_id: int):
response = await self._client.get(f"/apps/notes/api/v1/notes/{note_id}")
response.raise_for_status()
return response.json()
async def notes_create_note(
self,
*,
title: str | None = None,
content: str | None = None,
category: str | None = None,
):
body = {}
if title:
body.update({"title": title})
if content:
body.update({"content": content})
if category:
body.update({"category": category})
response = await self._client.post(
url="/apps/notes/api/v1/notes",
json=body,
)
response.raise_for_status()
return response.json()
async def notes_update_note(
self,
*,
note_id: int,
etag: str,
title: str | None = None,
content: str | None = None,
category: str | None = None,
):
# First, get the current note details to check for category change
old_note = None
try:
if category is not None: # Only fetch if category might change
old_note = await self.notes_get_note(note_id=note_id)
old_category = old_note.get("category", "")
logger.info(f"Current category for note {note_id}: '{old_category}'")
except Exception as e:
logger.warning(
f"Could not fetch current note {note_id} details before update: {e}"
)
# Continue with update even if we couldn't fetch current details
old_note = None
# Prepare update body
body = {}
if title:
body.update({"title": title})
if content:
body.update({"content": content})
if category:
body.update({"category": category})
logger.info(
"Attempting to update note %s with etag %s. Body: %s",
note_id,
etag,
body,
)
# Ensure conditional PUT using If-Match header is active
response = await self._client.put(
url=f"/apps/notes/api/v1/notes/{note_id}",
json=body,
headers={"If-Match": f'"{etag}"'},
)
logger.info(
"Update response for note %s: Status %s, Headers %s",
note_id,
response.status_code,
response.headers,
)
response.raise_for_status()
updated_note = response.json()
# Check for category change and clean up old attachment directory if needed
if (
old_note
and category is not None
and old_note.get("category", "") != category
):
logger.info(
f"Category changed from '{old_note.get('category', '')}' to '{category}' - cleaning up old attachment directory"
)
try:
await self._cleanup_old_attachment_directory(
note_id=note_id, old_category=old_note.get("category", "")
)
except Exception as e:
logger.error(
f"Error cleaning up old attachment directory for note {note_id}: {e}"
)
# Continue with update even if cleanup failed
return updated_note
async def notes_append_content(self, *, note_id: int, content: str):
"""Append content to an existing note.
The content will be separated by a newline and a delimiter `---`, so
one will not be required in the content provided to this tool
"""
logger.info(f"Appending content to note {note_id}")
# Get current note
current_note = await self.notes_get_note(note_id=note_id)
# Use fixed separator for consistency
separator = "\n---\n"
# Combine content
existing_content = current_note.get("content", "")
if existing_content:
new_content = existing_content + separator + content
else:
new_content = content # No separator needed for empty notes
logger.info(
f"Combining existing content ({len(existing_content)} chars) with new content ({len(content)} chars)"
)
# Update with combined content
return await self.notes_update_note(
note_id=note_id,
etag=current_note["etag"],
content=new_content,
title=None, # Keep existing title
category=None, # Keep existing category
)
async def notes_search_notes(self, *, query: str):
"""
Search notes using token-based matching with relevance ranking.
Returns notes sorted by relevance score.
"""
all_notes = await self.notes_get_all()
search_results = []
# Process the query
query_tokens = self.process_query(query)
# If empty query after processing, return empty results
if not query_tokens:
return []
# Process and score each note
for note in all_notes:
title_tokens, content_tokens = self.process_note_content(note)
score = self.calculate_score(query_tokens, title_tokens, content_tokens)
# Only include notes with a non-zero score
if score >= 0.5:
search_results.append(
{
"id": note.get("id"),
"title": note.get("title"),
"category": note.get("category"),
"modified": note.get("modified"),
"_score": score, # Include score for sorting (optional field)
}
)
# Sort by score in descending order
search_results.sort(key=lambda x: x["_score"], reverse=True)
# Keep score field for debugging
# for result in search_results:
# if "_score" in result:
# del result["_score"]
return search_results
def process_query(self, query: str) -> list[str]:
"""
Tokenize and normalize the search query.
"""
# Convert to lowercase and split into tokens
tokens = query.lower().split()
# Filter out very short tokens (optional)
tokens = [token for token in tokens if len(token) > 1]
# Could add stop word removal here
return tokens
def process_note_content(self, note: dict) -> tuple[list[str], list[str]]:
"""
Tokenize and normalize note title and content.
"""
# Process title
title = note.get("title", "").lower()
title_tokens = title.split()
# Process content
content = note.get("content", "").lower()
content_tokens = content.split()
return title_tokens, content_tokens
def calculate_score(
self,
query_tokens: list[str],
title_tokens: list[str],
content_tokens: list[str],
) -> float:
"""
Calculate a relevance score for a note based on query tokens.
"""
# Constants for weighting
TITLE_WEIGHT = 3.0
CONTENT_WEIGHT = 1.0
score = 0.0
# Count matches in title
title_matches = sum(1 for qt in query_tokens if qt in title_tokens)
if query_tokens: # Avoid division by zero
title_match_ratio = title_matches / len(query_tokens)
score += TITLE_WEIGHT * title_match_ratio
# Count matches in content
content_matches = sum(1 for qt in query_tokens if qt in content_tokens)
if query_tokens: # Avoid division by zero
content_match_ratio = content_matches / len(query_tokens)
score += CONTENT_WEIGHT * content_match_ratio
# If no tokens matched at all, return zero
if title_matches == 0 and content_matches == 0:
return 0.0
return score
async def _cleanup_old_attachment_directory(
self, *, note_id: int, old_category: str
):
"""
Clean up the attachment directory for a note in its old category location.
Called after a category change to prevent orphaned directories.
"""
# Construct path to old attachment directory
old_category_path_part = f"{old_category}/" if old_category else ""
old_attachment_dir_path = (
f"Notes/{old_category_path_part}.attachments.{note_id}/"
)
logger.info(f"Cleaning up old attachment directory: {old_attachment_dir_path}")
try:
delete_result = await self.delete_webdav_resource(
path=old_attachment_dir_path
)
logger.info(f"Cleanup of old attachment directory result: {delete_result}")
return delete_result
except Exception as e:
logger.error(f"Error during cleanup of old attachment directory: {e}")
raise e
async def delete_webdav_resource(self, *, path: str):
"""Delete a resource (file or directory) via WebDAV DELETE."""
# Ensure path ends with a slash if it's a directory
if not path.endswith("/"):
# This is a heuristic; a more robust solution would check resource type first
# but for the specific case of deleting the attachment directory, this is acceptable.
path_with_slash = f"{path}/"
else:
path_with_slash = path
webdav_path = f"{self._get_webdav_base_path()}/{path_with_slash.lstrip('/')}"
logger.info("Deleting WebDAV resource: %s", webdav_path)
headers = {"OCS-APIRequest": "true"}
try:
# First try a PROPFIND to verify resource exists
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try:
propfind_resp = await self._client.request(
"PROPFIND", webdav_path, headers=propfind_headers
)
logger.info(
f"Resource exists check (PROPFIND) status: {propfind_resp.status_code}"
)
# If we get here with 2xx, the resource exists
except HTTPStatusError as e:
if e.response.status_code == 404:
logger.info(
f"Resource '{webdav_path}' doesn't exist, no deletion needed."
)
return {"status_code": 404}
# For other errors, continue with deletion attempt
# Proceed with deletion
response = await self._client.delete(webdav_path, headers=headers)
response.raise_for_status() # Raises for 4xx/5xx status codes
logger.info(
"Successfully deleted WebDAV resource '%s' (Status: %s)",
webdav_path,
response.status_code,
)
# DELETE typically returns 204 No Content on success
return {"status_code": response.status_code}
except HTTPStatusError as e:
logger.warning(
"HTTP error deleting WebDAV resource '%s': %s",
webdav_path,
e,
)
# It's expected to get a 404 if the resource doesn't exist, which is fine.
# We only re-raise if it's not a 404.
if e.response.status_code != 404:
raise e
else:
logger.info("Resource '%s' not found, no deletion needed.", webdav_path)
return {"status_code": 404} # Indicate resource was not found
except Exception as e:
logger.warning(
"Unexpected error deleting WebDAV resource '%s': %s",
webdav_path,
e,
)
raise e
async def notes_delete_note(self, *, note_id: int):
"""Deletes a note via API and attempts to delete its attachment directory via WebDAV."""
# Fetch note details first to get the category for path construction
try:
note_details = await self.notes_get_note(note_id=note_id)
category = note_details.get("category", "")
# Check for other potential categories (if any note was moved between categories)
# We can't reliably detect this without a dedicated tracking mechanism, but we can
# implement a basic check for common category names and empty category
potential_categories = []
if category:
potential_categories.append(category) # Current category first
# Add empty category (uncategorized notes)
if category != "":
potential_categories.append("")
# We could add logic here to check for other common categories if needed
logger.info(
f"Note {note_id} has category: '{category}', will check attachment directories in: {potential_categories}"
)
except HTTPStatusError as e:
# If note doesn't exist (404), we can't delete attachments anyway.
# Re-raise other errors.
if e.response.status_code == 404:
logger.warning(
f"Note {note_id} not found when attempting delete. Skipping attachment cleanup."
)
# Still raise the 404 as the primary delete operation failed
raise e
else:
logger.error(
f"Error fetching note {note_id} details before deleting attachments: {e}"
)
raise e # Re-raise unexpected errors during fetch
# Proceed with API note deletion
logger.info(f"Deleting note {note_id} via API.")
response = await self._client.delete(f"/apps/notes/api/v1/notes/{note_id}")
response.raise_for_status() # Raise if API deletion fails
logger.info(f"Note {note_id} deleted successfully via API.")
json_response = response.json() # Usually empty on success
# Now, attempt to delete the associated attachments directory via WebDAV for each potential category
for cat in potential_categories:
cat_path_part = f"{cat}/" if cat else ""
attachment_dir_path = f"Notes/{cat_path_part}.attachments.{note_id}/"
logger.info(
f"Attempting to delete attachment directory for note {note_id} in category '{cat}' via WebDAV: {attachment_dir_path}"
)
try:
# delete_webdav_resource expects path relative to user's files dir
delete_result = await self.delete_webdav_resource(
path=attachment_dir_path
)
logger.info(
f"WebDAV deletion for category '{cat}' attachment directory: {delete_result}"
)
except Exception as e:
# Log the error but don't re-raise, as API note deletion itself was successful
# Also, we want to try other potential categories even if one fails
logger.warning(
f"Failed during WebDAV deletion for category '{cat}' attachment directory: {e}"
)
return json_response
# Removed incorrect get_note_attachment method that used Notes API
def _get_webdav_base_path(self) -> str:
"""Helper to get the base WebDAV path for the authenticated user."""
# Use the stored username
return f"/remote.php/dav/files/{self.username}"
# Removed _get_note_attachment_webdav_path helper
async def add_note_attachment(
self,
*,
note_id: int,
filename: str,
content: bytes,
category: str | None = None,
mime_type: str | None = None,
):
"""
Add/Update an attachment to a note via WebDAV PUT.
Requires the caller to provide the note's category.
"""
# Construct paths based on provided category
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
attachment_dir_segment = f".attachments.{note_id}"
parent_dir_webdav_rel_path = (
f"Notes/{category_path_part}{attachment_dir_segment}"
)
parent_dir_path = (
f"{webdav_base}/{parent_dir_webdav_rel_path}" # Full path for MKCOL
)
attachment_path = f"{parent_dir_path}/{filename}" # Full path for PUT
logger.info(
f"Uploading attachment for note {note_id} (category: '{category or ''}') to WebDAV path: {attachment_path}"
)
# Log current auth settings to diagnose the issue
logger.info(
"WebDAV auth settings - Username: %s, Auth Type: %s",
self.username,
type(self._client.auth).__name__,
)
if not mime_type:
mime_type, _ = mimetypes.guess_type(filename)
if not mime_type:
mime_type = "application/octet-stream" # Default if guessing fails
headers = {"Content-Type": mime_type, "OCS-APIRequest": "true"}
try:
# First check if we can access WebDAV at all with current credentials
# by checking the Notes directory
notes_dir_path = f"{webdav_base}/Notes"
logger.info("Testing WebDAV access to Notes directory: %s", notes_dir_path)
# Log details of the auth being used by the client for this specific request
if self._client.auth:
auth_header = (
self._client.auth.auth_flow(
self._client.build_request("GET", notes_dir_path)
)
.__next__()
.headers.get("Authorization")
)
logger.info(
"Authorization header for PROPFIND (Notes dir): %s",
(
auth_header
if auth_header
else "Not present or generated by auth flow"
),
)
else:
logger.info(
"No httpx.Auth object configured on the client for PROPFIND (Notes dir)."
)
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
logger.info("Headers for PROPFIND (Notes dir): %s", propfind_headers)
notes_dir_response = await self._client.request(
"PROPFIND", notes_dir_path, headers=propfind_headers
)
if notes_dir_response.status_code == 401:
logger.error(
"WebDAV authentication failed for Notes directory. Please verify WebDAV permissions."
)
raise HTTPStatusError(
f"Authentication error accessing WebDAV Notes directory: {notes_dir_response.status_code}",
request=notes_dir_response.request,
response=notes_dir_response,
)
elif notes_dir_response.status_code >= 400:
logger.error(
"Error accessing WebDAV Notes directory: %s",
notes_dir_response.status_code,
)
notes_dir_response.raise_for_status()
else:
logger.info(
"Successfully accessed WebDAV Notes directory (Status: %s)",
notes_dir_response.status_code,
)
# Ensure the parent directory exists using MKCOL
# parent_dir_path is now determined by the helper method
logger.info("Ensuring attachments directory exists: %s", parent_dir_path)
mkcol_headers = {"OCS-APIRequest": "true"}
logger.info("Headers for MKCOL (Attachments dir): %s", mkcol_headers)
mkcol_response = await self._client.request(
"MKCOL", parent_dir_path, headers=mkcol_headers
)
# MKCOL should return 201 Created or 405 Method Not Allowed (if directory already exists)
# We can ignore 405, but raise for other errors
if mkcol_response.status_code not in [201, 405]:
logger.warning(
"Unexpected status code %s when creating attachments directory",
mkcol_response.status_code,
)
mkcol_response.raise_for_status()
else:
logger.info(
"Created/verified directory: %s (Status: %s)",
parent_dir_path,
mkcol_response.status_code,
)
# Proceed with the PUT request
logger.info("Putting attachment file to: %s", attachment_path)
response = await self._client.put(
attachment_path, content=content, headers=headers
)
response.raise_for_status() # Raises for 4xx/5xx status codes
logger.info(
"Successfully uploaded attachment '%s' to note %s (Status: %s)",
filename,
note_id,
response.status_code,
)
# PUT typically returns 201 Created or 204 No Content on success
return {
"status_code": response.status_code
} # Return status or relevant info
except HTTPStatusError as e:
logger.error(
"HTTP error uploading attachment '%s' to note %s: %s",
filename,
note_id,
e,
)
raise e
except Exception as e:
logger.error(
"Unexpected error uploading attachment '%s' to note %s: %s",
filename,
note_id,
e,
)
raise e
async def get_note_attachment(
self, *, note_id: int, filename: str, category: str | None = None
):
"""
Fetch a specific attachment from a note via WebDAV GET.
Requires the caller to provide the note's category.
"""
# Construct path based on provided category
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
attachment_dir_segment = f".attachments.{note_id}"
attachment_path = f"{webdav_base}/Notes/{category_path_part}{attachment_dir_segment}/{filename}"
logger.info(
f"Fetching attachment for note {note_id} (category: '{category or ''}') from WebDAV path: {attachment_path}"
)
try:
response = await self._client.get(attachment_path)
response.raise_for_status()
content = response.content
mime_type = response.headers.get("content-type", "application/octet-stream")
logger.info(
"Successfully fetched attachment '%s' (%s, %d bytes)",
filename,
mime_type,
len(content),
)
return content, mime_type
except HTTPStatusError as e:
logger.error(
"HTTP error fetching attachment '%s' for note %s: %s",
filename,
note_id,
e,
)
raise e
except Exception as e:
logger.error(
"Unexpected error fetching attachment '%s' for note %s: %s",
filename,
note_id,
e,
)
raise e
+82
View File
@@ -0,0 +1,82 @@
import logging
import os
from httpx import AsyncClient, Auth, BasicAuth, Request, Response
from ..controllers.notes_search import NotesSearchController
from .calendar import CalendarClient
from .contacts import ContactsClient
from .notes import NotesClient
from .tables import TablesClient
from .webdav import WebDAVClient
logger = logging.getLogger(__name__)
def log_request(request: Request):
logger.info(
"Request event hook: %s %s - Waiting for content",
request.method,
request.url,
)
logger.info("Request body: %s", request.content)
logger.info("Headers: %s", request.headers)
def log_response(response: Response):
response.read() # Explicitly read the stream before accessing .text
logger.info("Response [%s] %s", response.status_code, response.text)
class NextcloudClient:
"""Main Nextcloud client that orchestrates all app clients."""
def __init__(self, base_url: str, username: str, auth: Auth | None = None):
self.username = username
self._client = AsyncClient(
base_url=base_url,
auth=auth,
# event_hooks={"request": [log_request], "response": [log_response]},
)
# Initialize app clients
self.notes = NotesClient(self._client, username)
self.webdav = WebDAVClient(self._client, username)
self.tables = TablesClient(self._client, username)
self.calendar = CalendarClient(self._client, username)
self.contacts = ContactsClient(self._client, username)
# Initialize controllers
self._notes_search = NotesSearchController()
@classmethod
def from_env(cls):
logger.info("Creating NC Client using env vars")
host = os.environ["NEXTCLOUD_HOST"]
username = os.environ["NEXTCLOUD_USERNAME"]
password = os.environ["NEXTCLOUD_PASSWORD"]
# Pass username to constructor
return cls(base_url=host, username=username, auth=BasicAuth(username, password))
async def capabilities(self):
response = await self._client.get(
"/ocs/v2.php/cloud/capabilities",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
)
response.raise_for_status()
return response.json()
async def notes_search_notes(self, *, query: str):
"""Search notes using token-based matching with relevance ranking."""
all_notes = await self.notes.get_all_notes()
return self._notes_search.search_notes(all_notes, query)
def _get_webdav_base_path(self) -> str:
"""Helper to get the base WebDAV path for the authenticated user."""
return f"/remote.php/dav/files/{self.username}"
async def close(self):
"""Close the HTTP client."""
await self._client.aclose()
+42
View File
@@ -0,0 +1,42 @@
"""Base client for Nextcloud operations with shared authentication."""
import logging
from abc import ABC
from httpx import AsyncClient
logger = logging.getLogger(__name__)
class BaseNextcloudClient(ABC):
"""Base class for all Nextcloud app clients."""
def __init__(self, http_client: AsyncClient, username: str):
"""Initialize with shared HTTP client and username.
Args:
http_client: Authenticated AsyncClient instance
username: Nextcloud username for WebDAV operations
"""
self._client = http_client
self.username = username
def _get_webdav_base_path(self) -> str:
"""Helper to get the base WebDAV path for the authenticated user."""
return f"/remote.php/dav/files/{self.username}"
async def _make_request(self, method: str, url: str, **kwargs):
"""Common request wrapper with logging and error handling.
Args:
method: HTTP method
url: Request URL
**kwargs: Additional request parameters
Returns:
Response object
"""
logger.debug(f"Making {method} request to {url}")
response = await self._client.request(method, url, **kwargs)
response.raise_for_status()
return response
+951
View File
@@ -0,0 +1,951 @@
"""CalDAV client for NextCloud calendar operations."""
import datetime as dt
import logging
import uuid
import xml.etree.ElementTree as ET
from typing import Any, Dict, List, Optional, Tuple
from httpx import HTTPStatusError
from icalendar import Alarm, Calendar
from icalendar import Event as ICalEvent
from icalendar import vRecur
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
class CalendarClient(BaseNextcloudClient):
"""Client for NextCloud CalDAV calendar operations."""
def _get_caldav_base_path(self) -> str:
"""Helper to get the base CalDAV path for calendars."""
return f"/remote.php/dav/calendars/{self.username}"
def _get_principals_path(self) -> str:
"""Helper to get the principals path for the user."""
return f"/remote.php/dav/principals/users/{self.username}"
async def list_calendars(self) -> List[Dict[str, Any]]:
"""List all available calendars for the user."""
caldav_path = self._get_caldav_base_path()
propfind_body = """<?xml version="1.0" encoding="utf-8"?>
<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav" xmlns:cs="http://calendarserver.org/ns/">
<d:prop>
<d:displayname/>
<d:resourcetype/>
<c:calendar-description/>
<cs:calendar-color/>
<c:supported-calendar-component-set/>
</d:prop>
</d:propfind>"""
headers = {
"Depth": "1",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"PROPFIND", caldav_path, content=propfind_body, headers=headers
)
# Parse XML response
root = ET.fromstring(response.content)
calendars = []
for response_elem in root.findall(".//{DAV:}response"):
href = response_elem.find(".//{DAV:}href")
if href is None:
continue
href_text = href.text or ""
if not href_text.endswith("/"):
continue # Skip non-calendar resources
# Extract calendar name from href
calendar_name = href_text.rstrip("/").split("/")[-1]
if not calendar_name or calendar_name == self.username:
continue
# Get properties
propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None:
continue
prop = propstat.find(".//{DAV:}prop")
if prop is None:
continue
# Check if it's a calendar resource
resourcetype = prop.find(".//{DAV:}resourcetype")
is_calendar = (
resourcetype is not None
and resourcetype.find(".//{urn:ietf:params:xml:ns:caldav}calendar")
is not None
)
if not is_calendar:
continue
# Extract calendar properties
displayname_elem = prop.find(".//{DAV:}displayname")
displayname = (
displayname_elem.text if displayname_elem is not None else calendar_name
)
description_elem = prop.find(
".//{urn:ietf:params:xml:ns:caldav}calendar-description"
)
description = description_elem.text if description_elem is not None else ""
color_elem = prop.find(".//{http://calendarserver.org/ns/}calendar-color")
color = color_elem.text if color_elem is not None else "#1976D2"
calendars.append(
{
"name": calendar_name,
"display_name": displayname,
"description": description,
"color": color,
"href": href_text,
}
)
logger.debug(f"Found {len(calendars)} calendars")
return calendars
async def get_calendar_events(
self,
calendar_name: str,
start_datetime: Optional[dt.datetime] = None,
end_datetime: Optional[dt.datetime] = None,
limit: int = 50,
) -> List[Dict[str, Any]]:
"""List events in a calendar within date range."""
calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
# Build time range filter if dates provided
time_range_filter = ""
if start_datetime or end_datetime:
# Convert datetime objects to CalDAV format (YYYYMMDDTHHMMSSZ)
start_dt = (
start_datetime.strftime("%Y%m%dT%H%M%SZ")
if start_datetime
else "19700101T000000Z"
)
end_dt = (
end_datetime.strftime("%Y%m%dT%H%M%SZ")
if end_datetime
else "20301231T235959Z"
)
time_range_filter = f"""
<c:time-range start="{start_dt}" end="{end_dt}"/>
"""
report_body = f"""<?xml version="1.0" encoding="utf-8"?>
<c:calendar-query xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">
<d:prop>
<d:getetag/>
<c:calendar-data/>
</d:prop>
<c:filter>
<c:comp-filter name="VCALENDAR">
<c:comp-filter name="VEVENT">
{time_range_filter}
</c:comp-filter>
</c:comp-filter>
</c:filter>
</c:calendar-query>"""
headers = {
"Depth": "1",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"REPORT", calendar_path, content=report_body, headers=headers
)
# Parse XML response and extract events
root = ET.fromstring(response.content)
events = []
for response_elem in root.findall(".//{DAV:}response"):
href = response_elem.find(".//{DAV:}href")
if href is None:
continue
propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None:
continue
prop = propstat.find(".//{DAV:}prop")
if prop is None:
continue
calendar_data = prop.find(".//{urn:ietf:params:xml:ns:caldav}calendar-data")
etag_elem = prop.find(".//{DAV:}getetag")
if calendar_data is not None and calendar_data.text:
event_data = self._parse_ical_event(calendar_data.text)
if event_data:
event_data["href"] = href.text
event_data["etag"] = etag_elem.text if etag_elem is not None else ""
events.append(event_data)
if len(events) >= limit:
break
logger.debug(f"Found {len(events)} events")
return events
async def create_event(
self, calendar_name: str, event_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Create a new calendar event with comprehensive features."""
event_uid = str(uuid.uuid4())
event_filename = f"{event_uid}.ics"
event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
# Create iCalendar event
ical_content = self._create_ical_event(event_data, event_uid)
headers = {
"Content-Type": "text/calendar; charset=utf-8",
"If-None-Match": "*", # Ensure we're creating, not updating
}
response = await self._make_request(
"PUT", event_path, content=ical_content, headers=headers
)
logger.debug(f"Created event {event_uid}")
return {
"uid": event_uid,
"href": event_path,
"etag": response.headers.get("etag", ""),
"status_code": response.status_code,
}
async def update_event(
self,
calendar_name: str,
event_uid: str,
event_data: Dict[str, Any],
etag: str = "",
) -> Dict[str, Any]:
"""Update an existing calendar event."""
event_filename = f"{event_uid}.ics"
event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
# Get existing event data to merge with updates
existing_event_data = {}
if not etag:
try:
existing_event_data, current_etag = await self.get_event(
calendar_name, event_uid
)
etag = current_etag
except Exception:
# Continue without etag if we can't get it
pass
# Merge existing data with new data (new data takes precedence)
merged_data = {**existing_event_data, **event_data}
# Create updated iCalendar event
ical_content = self._create_ical_event(merged_data, event_uid)
headers = {
"Content-Type": "text/calendar; charset=utf-8",
}
if etag:
headers["If-Match"] = etag
try:
response = await self._make_request(
"PUT", event_path, content=ical_content, headers=headers
)
logger.debug(f"Updated event {event_uid}")
return {
"uid": event_uid,
"href": event_path,
"etag": response.headers.get("etag", ""),
"status_code": response.status_code,
}
except HTTPStatusError as e:
logger.error(f"HTTP error updating event: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error updating event: {e}")
raise e
async def delete_event(self, calendar_name: str, event_uid: str) -> Dict[str, Any]:
"""Delete a calendar event."""
event_filename = f"{event_uid}.ics"
event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
try:
response = await self._make_request("DELETE", event_path)
logger.debug(f"Deleted event {event_uid}")
return {"status_code": response.status_code}
except HTTPStatusError as e:
if e.response.status_code == 404:
logger.debug(f"Event {event_uid} not found")
return {"status_code": 404}
logger.error(f"HTTP error deleting event: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error deleting event: {e}")
raise e
async def get_event(
self, calendar_name: str, event_uid: str
) -> Tuple[Dict[str, Any], str]:
"""Get detailed information about a specific event."""
event_filename = f"{event_uid}.ics"
event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
headers = {"Accept": "text/calendar"}
try:
response = await self._make_request("GET", event_path, headers=headers)
etag = response.headers.get("etag", "")
event_data = self._parse_ical_event(response.text)
if not event_data:
raise ValueError(f"Failed to parse event data for {event_uid}")
event_data["href"] = event_path
event_data["etag"] = etag
logger.debug(f"Retrieved event {event_uid}")
return event_data, etag
except HTTPStatusError as e:
logger.error(f"HTTP error getting event: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error getting event: {e}")
raise e
def _create_ical_event(self, event_data: Dict[str, Any], event_uid: str) -> str:
"""Create iCalendar content from event data."""
cal = Calendar()
cal.add("prodid", "-//NextCloud MCP Server//EN")
cal.add("version", "2.0")
event = ICalEvent()
event.add("uid", event_uid)
event.add("summary", event_data.get("title", ""))
event.add("description", event_data.get("description", ""))
event.add("location", event_data.get("location", ""))
# Handle dates/times
start_str = event_data.get("start_datetime", "")
end_str = event_data.get("end_datetime", "")
all_day = event_data.get("all_day", False)
if start_str: # Only parse if start_datetime is provided
if all_day:
start_date = dt.datetime.fromisoformat(start_str.split("T")[0]).date()
event.add("dtstart", start_date)
if end_str:
end_date = dt.datetime.fromisoformat(end_str.split("T")[0]).date()
event.add("dtend", end_date)
else:
start_dt = dt.datetime.fromisoformat(start_str.replace("Z", "+00:00"))
event.add("dtstart", start_dt)
if end_str:
end_dt = dt.datetime.fromisoformat(end_str.replace("Z", "+00:00"))
event.add("dtend", end_dt)
# Add categories
categories = event_data.get("categories", "")
if categories:
event.add("categories", categories.split(","))
# Add priority and status
priority = event_data.get("priority", 5)
event.add("priority", priority)
status = event_data.get("status", "CONFIRMED")
event.add("status", status)
# Add privacy classification
privacy = event_data.get("privacy", "PUBLIC")
event.add("class", privacy)
# Add URL
url = event_data.get("url", "")
if url:
event.add("url", url)
# Handle recurrence
recurring = event_data.get("recurring", False)
if recurring:
recurrence_rule = event_data.get("recurrence_rule", "")
if recurrence_rule:
event.add("rrule", vRecur.from_ical(recurrence_rule))
# Add alarms/reminders
reminder_minutes = event_data.get("reminder_minutes", 0)
if reminder_minutes > 0:
alarm = Alarm()
alarm.add("action", "DISPLAY")
alarm.add("description", "Event reminder")
alarm.add("trigger", dt.timedelta(minutes=-reminder_minutes))
event.add_component(alarm)
# Add attendees
attendees = event_data.get("attendees", "")
if attendees:
for email in attendees.split(","):
if email.strip():
event.add("attendee", f"mailto:{email.strip()}")
# Add timestamps
now = dt.datetime.now(dt.UTC)
event.add("created", now)
event.add("dtstamp", now)
event.add("last-modified", now)
cal.add_component(event)
return cal.to_ical().decode("utf-8")
def _parse_ical_event(self, ical_text: str) -> Optional[Dict[str, Any]]:
"""Parse iCalendar text and extract event data."""
try:
cal = Calendar.from_ical(ical_text)
for component in cal.walk():
if component.name == "VEVENT":
event_data = {
"uid": str(component.get("uid", "")),
"title": str(component.get("summary", "")),
"description": str(component.get("description", "")),
"location": str(component.get("location", "")),
"status": str(component.get("status", "CONFIRMED")),
"priority": int(component.get("priority", 5)),
"privacy": str(component.get("class", "PUBLIC")),
"url": str(component.get("url", "")),
}
# Handle dates
dtstart = component.get("dtstart")
if dtstart:
if isinstance(dtstart.dt, dt.date) and not isinstance(
dtstart.dt, dt.datetime
):
event_data["start_datetime"] = dtstart.dt.isoformat()
event_data["all_day"] = True
else:
event_data["start_datetime"] = dtstart.dt.isoformat()
event_data["all_day"] = False
dtend = component.get("dtend")
if dtend:
if isinstance(dtend.dt, dt.date) and not isinstance(
dtend.dt, dt.datetime
):
event_data["end_datetime"] = dtend.dt.isoformat()
else:
event_data["end_datetime"] = dtend.dt.isoformat()
# Handle categories
categories = component.get("categories")
if categories:
event_data["categories"] = self._extract_categories(categories)
# Handle recurrence
rrule = component.get("rrule")
if rrule:
event_data["recurring"] = True
event_data["recurrence_rule"] = str(rrule)
# Handle attendees
attendees = []
for attendee in component.get("attendee", []):
if isinstance(attendee, list):
attendees.extend(
str(a).replace("mailto:", "") for a in attendee
)
else:
attendees.append(str(attendee).replace("mailto:", ""))
if attendees:
event_data["attendees"] = ",".join(attendees)
return event_data
return None
except Exception as e:
logger.error(f"Error parsing iCalendar: {e}")
return None
def _extract_categories(self, categories_obj) -> str:
"""Extract categories from icalendar object to string."""
if not categories_obj:
return ""
try:
# Handle icalendar vCategory objects
if hasattr(categories_obj, "cats"):
# vCategory object has a 'cats' attribute that's a list
return ", ".join(str(cat) for cat in categories_obj.cats)
elif hasattr(categories_obj, "__iter__") and not isinstance(
categories_obj, str
):
# Handle lists or other iterables
return ", ".join(str(cat) for cat in categories_obj)
else:
# Handle strings or other objects
return str(categories_obj)
except Exception:
# Fallback to string conversion
return str(categories_obj)
async def search_events_across_calendars(
self,
start_datetime: Optional[dt.datetime] = None,
end_datetime: Optional[dt.datetime] = None,
filters: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
"""Search events across all calendars with advanced filtering."""
try:
calendars = await self.list_calendars()
all_events = []
for calendar in calendars:
try:
events = await self.get_calendar_events(
calendar["name"], start_datetime, end_datetime
)
# Apply filters if provided
if filters:
events = self._apply_event_filters(events, filters)
# Add calendar info to each event
for event in events:
event["calendar_name"] = calendar["name"]
event["calendar_display_name"] = calendar.get(
"display_name", calendar["name"]
)
all_events.extend(events)
except Exception as e:
logger.warning(
f"Error getting events from calendar {calendar['name']}: {e}"
)
continue
return all_events
except Exception as e:
logger.error(f"Error searching events across calendars: {e}")
raise
def _apply_event_filters(
self, events: List[Dict[str, Any]], filters: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Apply advanced filters to event list."""
filtered_events = []
for event in events:
# Skip if event doesn't match filters
if not self._event_matches_filters(event, filters):
continue
filtered_events.append(event)
return filtered_events
def _event_matches_filters(
self, event: Dict[str, Any], filters: Dict[str, Any]
) -> bool:
"""Check if an event matches the provided filters."""
try:
# Filter by minimum attendees
if "min_attendees" in filters:
attendees = event.get("attendees", "")
attendee_count = len(attendees.split(",")) if attendees else 0
if attendee_count < filters["min_attendees"]:
return False
# Filter by minimum duration
if "min_duration_minutes" in filters:
start_str = event.get("start_datetime", "")
end_str = event.get("end_datetime", "")
if start_str and end_str:
try:
start_dt = dt.datetime.fromisoformat(
start_str.replace("Z", "+00:00")
)
end_dt = dt.datetime.fromisoformat(
end_str.replace("Z", "+00:00")
)
duration_minutes = (end_dt - start_dt).total_seconds() / 60
if duration_minutes < filters["min_duration_minutes"]:
return False
except Exception:
pass
# Filter by categories
if "categories" in filters:
event_categories = event.get("categories", "").lower()
required_categories = [cat.lower() for cat in filters["categories"]]
if not any(cat in event_categories for cat in required_categories):
return False
# Filter by status
if "status" in filters:
if event.get("status", "").upper() != filters["status"].upper():
return False
# Filter by title contains
if "title_contains" in filters:
title = event.get("title", "").lower()
search_term = filters["title_contains"].lower()
if search_term not in title:
return False
# Filter by location contains
if "location_contains" in filters:
location = event.get("location", "").lower()
search_term = filters["location_contains"].lower()
if search_term not in location:
return False
return True
except Exception:
# If filtering fails, include the event
return True
async def find_availability(
self,
duration_minutes: int,
attendees: Optional[List[str]] = None,
start_datetime: Optional[dt.datetime] = None,
end_datetime: Optional[dt.datetime] = None,
constraints: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
"""Find available time slots for scheduling."""
try:
# Set default date range if not provided
if not start_datetime:
start_datetime = dt.datetime.now()
if not end_datetime:
end_datetime = dt.datetime.now() + dt.timedelta(days=7)
# Get all events in the date range
busy_events = await self.search_events_across_calendars(
start_datetime=start_datetime, end_datetime=end_datetime
)
# Filter events for relevant attendees if specified
if attendees:
relevant_events = []
for event in busy_events:
event_attendees = event.get("attendees", "").lower()
if any(
attendee.lower() in event_attendees for attendee in attendees
):
relevant_events.append(event)
busy_events = relevant_events
# Apply constraints
constraints = constraints or {}
business_hours_only = constraints.get("business_hours_only", False)
exclude_weekends = constraints.get("exclude_weekends", False)
preferred_times = constraints.get("preferred_times", [])
# Generate time slots
available_slots = self._generate_available_slots(
busy_events,
duration_minutes,
start_datetime,
end_datetime,
business_hours_only,
exclude_weekends,
preferred_times,
)
return available_slots
except Exception as e:
logger.error(f"Error finding availability: {e}")
raise
def _generate_available_slots(
self,
busy_events: List[Dict[str, Any]],
duration_minutes: int,
start_datetime: dt.datetime,
end_datetime: dt.datetime,
business_hours_only: bool,
exclude_weekends: bool,
preferred_times: List[str],
) -> List[Dict[str, Any]]:
"""Generate available time slots."""
available_slots = []
try:
current_date = start_datetime.replace(
hour=0, minute=0, second=0, microsecond=0
)
end_date_dt = end_datetime.replace(
hour=23, minute=59, second=59, microsecond=999999
)
while current_date <= end_date_dt:
# Skip weekends if requested
if exclude_weekends and current_date.weekday() >= 5:
current_date += dt.timedelta(days=1)
continue
# Generate slots for this day
day_slots = self._generate_day_slots(
current_date,
busy_events,
duration_minutes,
business_hours_only,
preferred_times,
)
available_slots.extend(day_slots)
current_date += dt.timedelta(days=1)
return available_slots[:10] # Limit to 10 slots
except Exception as e:
logger.error(f"Error generating available slots: {e}")
return []
def _generate_day_slots(
self,
date: dt.datetime,
busy_events: List[Dict[str, Any]],
duration_minutes: int,
business_hours_only: bool,
preferred_times: List[str],
) -> List[Dict[str, Any]]:
"""Generate available slots for a specific day."""
slots = []
try:
# Define working hours
if business_hours_only:
start_hour, end_hour = 9, 17
else:
start_hour, end_hour = 8, 20
# Get busy periods for this day
day_busy_periods = []
for event in busy_events:
try:
event_start = dt.datetime.fromisoformat(
event["start_datetime"].replace("Z", "+00:00")
)
event_end = dt.datetime.fromisoformat(
event["end_datetime"].replace("Z", "+00:00")
)
# Check if event is on this day
if event_start.date() == date.date():
day_busy_periods.append((event_start.time(), event_end.time()))
except Exception:
continue
# Sort busy periods
day_busy_periods.sort()
# Generate potential slots
current_time = date.replace(
hour=start_hour, minute=0, second=0, microsecond=0
)
end_time = date.replace(hour=end_hour, minute=0, second=0, microsecond=0)
slot_duration = dt.timedelta(minutes=duration_minutes)
while current_time + slot_duration <= end_time:
slot_end = current_time + slot_duration
# Check if slot conflicts with any busy period
if not self._slot_conflicts(
current_time.time(), slot_end.time(), day_busy_periods
):
# Check preferred times if specified
if not preferred_times or self._slot_in_preferred_times(
current_time.time(), preferred_times
):
slots.append(
{
"start_datetime": current_time.isoformat(),
"end_datetime": slot_end.isoformat(),
"duration_minutes": duration_minutes,
"date": date.date().isoformat(),
}
)
current_time += dt.timedelta(minutes=30) # 30-minute increments
return slots
except Exception as e:
logger.error(f"Error generating day slots: {e}")
return []
def _slot_conflicts(self, slot_start, slot_end, busy_periods):
"""Check if a time slot conflicts with busy periods."""
for busy_start, busy_end in busy_periods:
if slot_start < busy_end and slot_end > busy_start:
return True
return False
def _slot_in_preferred_times(self, slot_start, preferred_times):
"""Check if slot falls within preferred time ranges."""
if not preferred_times:
return True
for time_range in preferred_times:
try:
start_str, end_str = time_range.split("-")
pref_start = dt.datetime.strptime(start_str, "%H:%M").time()
pref_end = dt.datetime.strptime(end_str, "%H:%M").time()
if pref_start <= slot_start <= pref_end:
return True
except Exception:
continue
return False
async def bulk_update_events(
self, filter_criteria: Dict[str, Any], update_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Bulk update events matching filter criteria."""
try:
# Convert string dates to datetime objects if present
start_datetime = None
end_datetime = None
if "start_date" in filter_criteria and filter_criteria["start_date"]:
start_datetime = dt.datetime.fromisoformat(
filter_criteria["start_date"]
)
if "end_date" in filter_criteria and filter_criteria["end_date"]:
end_datetime = dt.datetime.fromisoformat(filter_criteria["end_date"])
# Find events matching criteria
events = await self.search_events_across_calendars(
start_datetime=start_datetime,
end_datetime=end_datetime,
filters=filter_criteria,
)
updated_count = 0
failed_count = 0
results = []
for event in events:
try:
# Update the event
await self.update_event(
event["calendar_name"], event["uid"], update_data
)
updated_count += 1
results.append(
{
"uid": event["uid"],
"status": "updated",
"title": event.get("title", ""),
}
)
except Exception as e:
failed_count += 1
results.append(
{
"uid": event["uid"],
"status": "failed",
"error": str(e),
"title": event.get("title", ""),
}
)
return {
"total_found": len(events),
"updated_count": updated_count,
"failed_count": failed_count,
"results": results,
}
except Exception as e:
logger.error(f"Error in bulk update: {e}")
raise
async def create_calendar(
self,
calendar_name: str,
display_name: str = "",
description: str = "",
color: str = "#1976D2",
) -> Dict[str, Any]:
"""Create a new calendar."""
try:
# Calendar creation via CalDAV MKCALENDAR
calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
# Create MKCALENDAR body
mkcol_body = f"""<?xml version="1.0" encoding="utf-8"?>
<mkcalendar xmlns="urn:ietf:params:xml:ns:caldav" xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
<d:set>
<d:prop>
<d:displayname>{display_name or calendar_name}</d:displayname>
<cs:calendar-color>{color}</cs:calendar-color>
<caldav:calendar-description xmlns:caldav="urn:ietf:params:xml:ns:caldav">{description}</caldav:calendar-description>
<caldav:supported-calendar-component-set xmlns:caldav="urn:ietf:params:xml:ns:caldav">
<caldav:comp name="VEVENT"/>
</caldav:supported-calendar-component-set>
</d:prop>
</d:set>
</mkcalendar>"""
headers = {"Content-Type": "application/xml", "Depth": "0"}
response = await self._make_request(
"MKCALENDAR", calendar_path, content=mkcol_body, headers=headers
)
logger.debug(f"Created calendar: {calendar_name}")
return {
"name": calendar_name,
"display_name": display_name or calendar_name,
"description": description,
"color": color,
"status_code": response.status_code,
}
except Exception as e:
logger.error(f"Error creating calendar {calendar_name}: {e}")
raise
async def delete_calendar(self, calendar_name: str) -> Dict[str, Any]:
"""Delete a calendar."""
try:
calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
response = await self._make_request("DELETE", calendar_path)
logger.debug(f"Deleted calendar: {calendar_name}")
return {"status_code": response.status_code}
except Exception as e:
logger.error(f"Error deleting calendar {calendar_name}: {e}")
raise
+235
View File
@@ -0,0 +1,235 @@
"""CardDAV client for NextCloud contacts operations."""
import logging
from .base import BaseNextcloudClient
import xml.etree.ElementTree as ET
from pythonvCard4.vcard import Contact
logger = logging.getLogger(__name__)
class ContactsClient(BaseNextcloudClient):
"""Client for NextCloud CardDAV contact operations."""
def _get_carddav_base_path(self) -> str:
"""Helper to get the base CardDAV path for contacts."""
return f"/remote.php/dav/addressbooks/users/{self.username}"
async def list_addressbooks(self):
"""List all available addressbooks for the user."""
carddav_path = self._get_carddav_base_path()
propfind_body = """<?xml version="1.0" encoding="utf-8"?>
<d:propfind xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
<d:prop>
<d:displayname/>
<d:getctag />
</d:prop>
</d:propfind>"""
headers = {
# "Depth": "0",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"PROPFIND", carddav_path, content=propfind_body, headers=headers
)
ns = {"d": "DAV:"}
# logger.info(response.content)
root = ET.fromstring(response.content)
addressbooks = []
for response_elem in root.findall(".//d:response", ns):
href = response_elem.find(".//d:href", ns)
if href is None:
continue
href_text = href.text or ""
if not href_text.endswith("/"):
continue # Skip non-addressbook resources
# Extract addressbook name from href
addressbook_name = href_text.rstrip("/").split("/")[-1]
if not addressbook_name or addressbook_name == self.username:
continue
# Get properties
propstat = response_elem.find(".//d:propstat", ns)
if propstat is None:
continue
prop = propstat.find(".//d:prop", ns)
if prop is None:
continue
displayname_elem = prop.find(".//d:displayname", ns)
displayname = (
displayname_elem.text
if displayname_elem is not None
else addressbook_name
)
getctag_elem = prop.find(".//d:getctag", ns)
getctag = getctag_elem.text if getctag_elem is not None else None
addressbooks.append(
{
"name": addressbook_name,
"display_name": displayname,
"getctag": getctag,
}
)
logger.debug(f"Found {len(addressbooks)} addressbooks")
return addressbooks
async def create_addressbook(self, *, name: str, display_name: str):
"""Create a new addressbook."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{name}/"
prop_body = f"""<?xml version="1.0" encoding="utf-8"?>
<d:mkcol xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:carddav">
<d:set>
<d:prop>
<d:resourcetype>
<d:collection/>
<c:addressbook/>
</d:resourcetype>
<d:displayname>{display_name}</d:displayname>
</d:prop>
</d:set>
</d:mkcol>"""
headers = {
"Content-Type": "application/xml",
}
await self._make_request("MKCOL", url, content=prop_body, headers=headers)
async def delete_addressbook(self, *, name: str):
"""Delete an addressbook."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{name}/"
await self._make_request("DELETE", url)
async def create_contact(self, *, addressbook: str, uid: str, contact_data: dict):
"""Create a new contact."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
contact = Contact(fn=contact_data.get("fn"), uid=uid)
if "email" in contact_data:
contact.email = [{"value": contact_data["email"], "type": ["HOME"]}]
if "tel" in contact_data:
contact.tel = [{"value": contact_data["tel"], "type": ["HOME"]}]
vcard = contact.to_vcard()
headers = {
"Content-Type": "text/vcard; charset=utf-8",
"If-None-Match": "*",
}
await self._make_request("PUT", url, content=vcard, headers=headers)
async def delete_contact(self, *, addressbook: str, uid: str):
"""Delete a contact."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
await self._make_request("DELETE", url)
async def list_contacts(self, *, addressbook: str):
"""List all available contacts for addressbook."""
carddav_path = self._get_carddav_base_path()
report_body = """<?xml version="1.0" encoding="utf-8"?>
<card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
<d:prop>
<d:getetag />
<card:address-data />
</d:prop>
</card:addressbook-query>"""
headers = {
"Depth": "1",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"REPORT",
f"{carddav_path}/{addressbook}",
content=report_body,
headers=headers,
)
ns = {"d": "DAV:", "card": "urn:ietf:params:xml:ns:carddav"}
# logger.info(response.text)
root = ET.fromstring(response.content)
contacts = []
for response_elem in root.findall(".//d:response", ns):
href = response_elem.find(".//d:href", ns)
if href is None:
logger.info("Skip missing href")
continue
href_text = href.text or ""
# logger.info("Href text: %s", href_text)
# if not href_text.endswith("/"):
# logger.info("# Skip non-addressbook resources")
# continue
# Extract vcard id from href
vcard_id = href_text.rstrip("/").split("/")[-1]
if not vcard_id:
logger.info("Skip missing vcard_id")
continue
vcard_id = vcard_id.replace(".vcf", "")
# Get properties
propstat = response_elem.find(".//d:propstat", ns)
if propstat is None:
logger.info("Skip missing propstat")
continue
prop = propstat.find(".//d:prop", ns)
if prop is None:
logger.info("Skip missing prop")
continue
getetag_elem = prop.find(".//d:getetag", ns)
getetag = getetag_elem.text if getetag_elem is not None else None
addressdata_elem = prop.find(".//card:address-data", ns)
addressdata = (
addressdata_elem.text if addressdata_elem is not None else None
)
if addressdata is None:
logger.info("Skip missing addressdata")
continue
contact = Contact.from_vcard(addressdata)
contacts.append(
{
"vcard_id": vcard_id,
"getetag": getetag,
"contact": {
"fullname": contact.fn,
"nickname": contact.nickname,
"birthday": contact.bday,
"email": contact.email,
},
"addressdata": addressdata,
}
)
logger.debug(f"Found {len(contacts)} contacts")
return contacts
+199
View File
@@ -0,0 +1,199 @@
"""Client for Nextcloud Notes app operations."""
import logging
from typing import Any, Dict, List, Optional
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
class NotesClient(BaseNextcloudClient):
"""Client for Nextcloud Notes app operations."""
async def get_settings(self) -> Dict[str, Any]:
"""Get Notes app settings."""
response = await self._make_request("GET", "/apps/notes/api/v1/settings")
return response.json()
async def get_all_notes(self) -> List[Dict[str, Any]]:
"""Get all notes."""
response = await self._make_request("GET", "/apps/notes/api/v1/notes")
return response.json()
async def get_note(self, note_id: int) -> Dict[str, Any]:
"""Get a specific note by ID."""
response = await self._make_request(
"GET", f"/apps/notes/api/v1/notes/{note_id}"
)
return response.json()
async def create_note(
self,
title: Optional[str] = None,
content: Optional[str] = None,
category: Optional[str] = None,
) -> Dict[str, Any]:
"""Create a new note."""
body = {}
if title:
body["title"] = title
if content:
body["content"] = content
if category:
body["category"] = category
response = await self._make_request(
"POST", "/apps/notes/api/v1/notes", json=body
)
return response.json()
async def update(
self,
note_id: int,
etag: str,
title: Optional[str] = None,
content: Optional[str] = None,
category: Optional[str] = None,
) -> Dict[str, Any]:
"""Update an existing note."""
# Get current note details to check for category change
old_note = None
try:
if category is not None:
old_note = await self.get_note(note_id)
old_category = old_note.get("category", "")
logger.info(f"Current category for note {note_id}: '{old_category}'")
except Exception as e:
logger.warning(
f"Could not fetch current note {note_id} details before update: {e}"
)
old_note = None
# Prepare update body
body = {}
if title:
body["title"] = title
if content:
body["content"] = content
if category:
body["category"] = category
logger.info(
f"Attempting to update note {note_id} with etag {etag}. Body: {body}"
)
response = await self._make_request(
"PUT",
f"/apps/notes/api/v1/notes/{note_id}",
json=body,
headers={"If-Match": f'"{etag}"'},
)
logger.info(
f"Update response for note {note_id}: Status {response.status_code}"
)
updated_note = response.json()
# Check for category change and cleanup old attachment directory if needed
if (
old_note
and category is not None
and old_note.get("category", "") != category
):
logger.info(
f"Category changed from '{old_note.get('category', '')}' to '{category}' - cleaning up old attachment directory"
)
try:
# Import here to avoid circular imports
from .webdav import WebDAVClient
webdav_client = WebDAVClient(self._client, self.username)
await webdav_client.cleanup_old_attachment_directory(
note_id=note_id, old_category=old_note.get("category", "")
)
except Exception as e:
logger.error(
f"Error cleaning up old attachment directory for note {note_id}: {e}"
)
return updated_note
async def delete_note(self, note_id: int) -> Dict[str, Any]:
"""Delete a note and its attachments."""
# Fetch note details first to get category for cleanup
try:
note_details = await self.get_note(note_id)
category = note_details.get("category", "")
# Determine potential categories for cleanup
potential_categories = []
if category:
potential_categories.append(category)
if category != "":
potential_categories.append("") # Empty category
logger.info(
f"Note {note_id} has category: '{category}', will check attachment directories in: {potential_categories}"
)
except Exception as e:
logger.warning(
f"Could not fetch note {note_id} details before deletion: {e}"
)
potential_categories = ["", "Unknown"] # Try common categories
# Delete the note via API
logger.info(f"Deleting note {note_id} via API")
response = await self._make_request(
"DELETE", f"/apps/notes/api/v1/notes/{note_id}"
)
logger.info(f"Note {note_id} deleted successfully via API")
json_response = response.json()
# Clean up attachment directories
try:
from .webdav import WebDAVClient
webdav_client = WebDAVClient(self._client, self.username)
for cat in potential_categories:
try:
await webdav_client.cleanup_note_attachments(note_id, cat)
except Exception as e:
logger.warning(
f"Failed to cleanup attachments for category '{cat}': {e}"
)
except Exception as e:
logger.warning(f"Error during attachment cleanup: {e}")
return json_response
async def append_content(self, note_id: int, content: str) -> Dict[str, Any]:
"""Append content to an existing note with a separator."""
logger.info(f"Appending content to note {note_id}")
# Get current note
current_note = await self.get_note(note_id)
# Use fixed separator for consistency
separator = "\n---\n"
# Combine content
existing_content = current_note.get("content", "")
if existing_content:
new_content = existing_content + separator + content
else:
new_content = content # No separator needed for empty notes
logger.info(
f"Combining existing content ({len(existing_content)} chars) with new content ({len(content)} chars)"
)
# Update with combined content
return await self.update(
note_id=note_id,
etag=current_note["etag"],
content=new_content,
title=None, # Keep existing title
category=None, # Keep existing category
)
+125
View File
@@ -0,0 +1,125 @@
"""Client for Nextcloud Tables app operations."""
import logging
from typing import Any, Dict, List, Optional
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
class TablesClient(BaseNextcloudClient):
"""Client for Nextcloud Tables app operations."""
async def list_tables(self) -> List[Dict[str, Any]]:
"""List all tables available to the user."""
response = await self._make_request(
"GET",
"/ocs/v2.php/apps/tables/api/2/tables",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
)
result = response.json()
return result["ocs"]["data"]
async def get_table_schema(self, table_id: int) -> Dict[str, Any]:
"""Get the schema/structure of a specific table including columns and views."""
# Using v1 API as v2 schema endpoint had issues during testing
response = await self._make_request(
"GET", f"/index.php/apps/tables/api/1/tables/{table_id}/scheme"
)
return response.json()
async def get_table_rows(
self, table_id: int, limit: Optional[int] = None, offset: Optional[int] = None
) -> List[Dict[str, Any]]:
"""Read rows from a table with optional pagination."""
params = {}
if limit is not None:
params["limit"] = limit
if offset is not None:
params["offset"] = offset
response = await self._make_request(
"GET", f"/index.php/apps/tables/api/1/tables/{table_id}/rows", params=params
)
return response.json()
async def create_row(self, table_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""Insert a new row into a table.
Args:
table_id: ID of the table to insert into
data: Dictionary mapping column IDs to values, e.g. {1: "text", 2: 42}
"""
# Transform data to API format: {"data": {"1": "text", "2": 42}}
api_data = {str(k): v for k, v in data.items()}
response = await self._make_request(
"POST",
f"/ocs/v2.php/apps/tables/api/2/tables/{table_id}/rows",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
json={"data": api_data},
)
result = response.json()
return result["ocs"]["data"]
async def update_row(self, row_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing row in a table.
Args:
row_id: ID of the row to update
data: Dictionary mapping column IDs to new values, e.g. {1: "new text", 2: 99}
"""
# Transform data to API format for v1 endpoint
api_data = {str(k): v for k, v in data.items()}
response = await self._make_request(
"PUT",
f"/index.php/apps/tables/api/1/rows/{row_id}",
json={"data": api_data},
)
return response.json()
async def delete_row(self, row_id: int) -> Dict[str, Any]:
"""Delete a row from a table."""
response = await self._make_request(
"DELETE", f"/index.php/apps/tables/api/1/rows/{row_id}"
)
return response.json()
def transform_row_data(
self, rows: List[Dict[str, Any]], columns: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""Transform raw row data into more readable format using column names.
Args:
rows: Raw row data from the API
columns: Column definitions from table schema
Returns:
List of rows with column names as keys instead of column IDs
"""
# Create mapping from column ID to column title
column_map = {col["id"]: col["title"] for col in columns}
transformed_rows = []
for row in rows:
transformed_row = {
"id": row["id"],
"tableId": row["tableId"],
"createdBy": row["createdBy"],
"createdAt": row["createdAt"],
"lastEditBy": row["lastEditBy"],
"lastEditAt": row["lastEditAt"],
"data": {},
}
# Transform data array to column_name: value mapping
for item in row["data"]:
column_id = item["columnId"]
column_name = column_map.get(column_id, f"column_{column_id}")
transformed_row["data"][column_name] = item["value"]
transformed_rows.append(transformed_row)
return transformed_rows
+418
View File
@@ -0,0 +1,418 @@
"""WebDAV client for Nextcloud file operations."""
import logging
import mimetypes
import xml.etree.ElementTree as ET
from typing import Any, Dict, List, Optional, Tuple
from httpx import HTTPStatusError
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
class WebDAVClient(BaseNextcloudClient):
"""Client for Nextcloud WebDAV operations."""
async def delete_resource(self, path: str) -> Dict[str, Any]:
"""Delete a resource (file or directory) via WebDAV DELETE."""
# Ensure path ends with a slash if it's a directory
if not path.endswith("/"):
path_with_slash = f"{path}/"
else:
path_with_slash = path
webdav_path = f"{self._get_webdav_base_path()}/{path_with_slash.lstrip('/')}"
logger.debug(f"Deleting WebDAV resource: {webdav_path}")
headers = {"OCS-APIRequest": "true"}
try:
# First try a PROPFIND to verify resource exists
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try:
propfind_resp = await self._client.request(
"PROPFIND", webdav_path, headers=propfind_headers
)
logger.debug(
f"Resource exists check status: {propfind_resp.status_code}"
)
except HTTPStatusError as e:
if e.response.status_code == 404:
logger.debug(f"Resource '{path}' doesn't exist, no deletion needed")
return {"status_code": 404}
# For other errors, continue with deletion attempt
# Proceed with deletion
response = await self._client.delete(webdav_path, headers=headers)
response.raise_for_status()
logger.debug(f"Successfully deleted WebDAV resource '{path}'")
return {"status_code": response.status_code}
except HTTPStatusError as e:
if e.response.status_code == 404:
logger.debug(f"Resource '{path}' not found, no deletion needed")
return {"status_code": 404}
else:
logger.error(f"HTTP error deleting WebDAV resource '{path}': {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error deleting WebDAV resource '{path}': {e}")
raise e
async def cleanup_old_attachment_directory(
self, note_id: int, old_category: str
) -> Dict[str, Any]:
"""Clean up the attachment directory for a note in its old category location."""
old_category_path_part = f"{old_category}/" if old_category else ""
old_attachment_dir_path = (
f"Notes/{old_category_path_part}.attachments.{note_id}/"
)
logger.debug(f"Cleaning up old attachment directory: {old_attachment_dir_path}")
try:
delete_result = await self.delete_resource(path=old_attachment_dir_path)
logger.debug(f"Cleanup result: {delete_result}")
return delete_result
except Exception as e:
logger.error(f"Error during cleanup of old attachment directory: {e}")
raise e
async def cleanup_note_attachments(
self, note_id: int, category: str
) -> Dict[str, Any]:
"""Clean up attachment directory for a specific note and category."""
cat_path_part = f"{category}/" if category else ""
attachment_dir_path = f"Notes/{cat_path_part}.attachments.{note_id}/"
logger.debug(
f"Cleaning up attachments for note {note_id} in category '{category}'"
)
try:
delete_result = await self.delete_resource(path=attachment_dir_path)
logger.debug(f"Cleanup result for note {note_id}: {delete_result}")
return delete_result
except Exception as e:
logger.error(f"Failed cleaning up attachments for note {note_id}: {e}")
raise e
async def add_note_attachment(
self,
note_id: int,
filename: str,
content: bytes,
category: Optional[str] = None,
mime_type: Optional[str] = None,
) -> Dict[str, Any]:
"""Add/Update an attachment to a note via WebDAV PUT."""
# Construct paths based on provided category
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
attachment_dir_segment = f".attachments.{note_id}"
parent_dir_webdav_rel_path = (
f"Notes/{category_path_part}{attachment_dir_segment}"
)
parent_dir_path = f"{webdav_base}/{parent_dir_webdav_rel_path}"
attachment_path = f"{parent_dir_path}/{filename}"
logger.debug(f"Uploading attachment '{filename}' for note {note_id}")
if not mime_type:
mime_type, _ = mimetypes.guess_type(filename)
if not mime_type:
mime_type = "application/octet-stream"
headers = {"Content-Type": mime_type, "OCS-APIRequest": "true"}
try:
# First check if we can access WebDAV at all
notes_dir_path = f"{webdav_base}/Notes"
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
notes_dir_response = await self._client.request(
"PROPFIND", notes_dir_path, headers=propfind_headers
)
if notes_dir_response.status_code == 401:
logger.error("WebDAV authentication failed for Notes directory")
raise HTTPStatusError(
f"Authentication error accessing WebDAV Notes directory: {notes_dir_response.status_code}",
request=notes_dir_response.request,
response=notes_dir_response,
)
elif notes_dir_response.status_code >= 400:
logger.error(
f"Error accessing WebDAV Notes directory: {notes_dir_response.status_code}"
)
notes_dir_response.raise_for_status()
# Ensure the parent directory exists using MKCOL
mkcol_headers = {"OCS-APIRequest": "true"}
mkcol_response = await self._client.request(
"MKCOL", parent_dir_path, headers=mkcol_headers
)
# MKCOL should return 201 Created or 405 Method Not Allowed (if directory already exists)
if mkcol_response.status_code not in [201, 405]:
logger.error(
f"Unexpected status code {mkcol_response.status_code} when creating attachments directory"
)
mkcol_response.raise_for_status()
# Proceed with the PUT request
response = await self._client.put(
attachment_path, content=content, headers=headers
)
response.raise_for_status()
logger.debug(
f"Successfully uploaded attachment '{filename}' to note {note_id}"
)
return {"status_code": response.status_code}
except HTTPStatusError as e:
logger.error(
f"HTTP error uploading attachment '{filename}' to note {note_id}: {e}"
)
raise e
except Exception as e:
logger.error(
f"Unexpected error uploading attachment '{filename}' to note {note_id}: {e}"
)
raise e
async def get_note_attachment(
self, note_id: int, filename: str, category: Optional[str] = None
) -> Tuple[bytes, str]:
"""Fetch a specific attachment from a note via WebDAV GET."""
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
attachment_dir_segment = f".attachments.{note_id}"
attachment_path = f"{webdav_base}/Notes/{category_path_part}{attachment_dir_segment}/{filename}"
logger.debug(f"Fetching attachment '{filename}' for note {note_id}")
try:
response = await self._client.get(attachment_path)
response.raise_for_status()
content = response.content
mime_type = response.headers.get("content-type", "application/octet-stream")
logger.debug(
f"Successfully fetched attachment '{filename}' ({len(content)} bytes)"
)
return content, mime_type
except HTTPStatusError as e:
if e.response.status_code == 404:
logger.debug(f"Attachment '{filename}' not found for note {note_id}")
else:
logger.error(
f"HTTP error fetching attachment '{filename}' for note {note_id}: {e}"
)
raise e
except Exception as e:
logger.error(
f"Unexpected error fetching attachment '{filename}' for note {note_id}: {e}"
)
raise e
async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
"""List files and directories in the specified path via WebDAV PROPFIND."""
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
if not webdav_path.endswith("/"):
webdav_path += "/"
logger.debug(f"Listing directory: {path}")
propfind_body = """<?xml version="1.0"?>
<d:propfind xmlns:d="DAV:">
<d:prop>
<d:displayname/>
<d:getcontentlength/>
<d:getcontenttype/>
<d:getlastmodified/>
<d:resourcetype/>
</d:prop>
</d:propfind>"""
headers = {"Depth": "1", "Content-Type": "text/xml", "OCS-APIRequest": "true"}
try:
response = await self._client.request(
"PROPFIND", webdav_path, content=propfind_body, headers=headers
)
response.raise_for_status()
# Parse the XML response
root = ET.fromstring(response.content)
items = []
# Skip the first response (the directory itself)
responses = root.findall(".//{DAV:}response")[1:]
for response_elem in responses:
href = response_elem.find(".//{DAV:}href")
if href is None:
continue
# Extract file/directory name from href
href_text = href.text or ""
name = href_text.rstrip("/").split("/")[-1]
if not name:
continue
# Get properties
propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None:
continue
prop = propstat.find(".//{DAV:}prop")
if prop is None:
continue
# Determine if it's a directory
resourcetype = prop.find(".//{DAV:}resourcetype")
is_directory = (
resourcetype is not None
and resourcetype.find(".//{DAV:}collection") is not None
)
# Get other properties
size_elem = prop.find(".//{DAV:}getcontentlength")
size = (
int(size_elem.text)
if size_elem is not None and size_elem.text
else 0
)
content_type_elem = prop.find(".//{DAV:}getcontenttype")
content_type = (
content_type_elem.text if content_type_elem is not None else None
)
modified_elem = prop.find(".//{DAV:}getlastmodified")
modified = modified_elem.text if modified_elem is not None else None
items.append(
{
"name": name,
"path": f"{path.rstrip('/')}/{name}" if path else name,
"is_directory": is_directory,
"size": size if not is_directory else None,
"content_type": content_type,
"last_modified": modified,
}
)
logger.debug(f"Found {len(items)} items in directory: {path}")
return items
except HTTPStatusError as e:
logger.error(f"HTTP error listing directory '{webdav_path}': {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error listing directory '{webdav_path}': {e}")
raise e
async def read_file(self, path: str) -> Tuple[bytes, str]:
"""Read a file's content via WebDAV GET."""
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
logger.debug(f"Reading file: {path}")
try:
response = await self._client.get(webdav_path)
response.raise_for_status()
content = response.content
content_type = response.headers.get(
"content-type", "application/octet-stream"
)
logger.debug(f"Successfully read file '{path}' ({len(content)} bytes)")
return content, content_type
except HTTPStatusError as e:
logger.error(f"HTTP error reading file '{path}': {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error reading file '{path}': {e}")
raise e
async def write_file(
self, path: str, content: bytes, content_type: Optional[str] = None
) -> Dict[str, Any]:
"""Write content to a file via WebDAV PUT."""
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
logger.debug(f"Writing file: {path}")
if not content_type:
content_type, _ = mimetypes.guess_type(path)
if not content_type:
content_type = "application/octet-stream"
headers = {"Content-Type": content_type, "OCS-APIRequest": "true"}
try:
response = await self._client.put(
webdav_path, content=content, headers=headers
)
response.raise_for_status()
logger.debug(f"Successfully wrote file '{path}'")
return {"status_code": response.status_code}
except HTTPStatusError as e:
logger.error(f"HTTP error writing file '{path}': {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error writing file '{path}': {e}")
raise e
async def create_directory(
self, path: str, recursive: bool = False
) -> Dict[str, Any]:
"""Create a directory via WebDAV MKCOL."""
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
if not webdav_path.endswith("/"):
webdav_path += "/"
logger.debug(f"Creating directory: {path}")
headers = {"OCS-APIRequest": "true"}
try:
response = await self._client.request("MKCOL", webdav_path, headers=headers)
response.raise_for_status()
logger.debug(f"Successfully created directory '{path}'")
return {"status_code": response.status_code}
except HTTPStatusError as e:
# Method Not Allowed - directory already exists
if e.response.status_code == 405:
logger.debug(f"Directory '{path}' already exists")
return {"status_code": 405, "message": "Directory already exists"}
# File Conflict - parent directory does not exist
if e.response.status_code == 409 and recursive:
# Extract parent directory path
path_parts = path.strip("/").split("/")
if len(path_parts) > 1:
parent_dir = "/".join(path_parts[:-1])
logger.debug(
f"Parent directory '{parent_dir}' doesn't exist, creating recursively"
)
await self.create_directory(parent_dir, recursive)
# Now try to create the original directory again
return await self.create_directory(path, recursive)
else:
# This shouldn't happen for single-level directories under root
logger.error(f"409 conflict for single-level directory '{path}'")
raise e
logger.error(f"HTTP error creating directory '{path}': {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error creating directory '{path}': {e}")
raise e
+2 -2
View File
@@ -21,12 +21,12 @@ LOGGING_CONFIG = {
},
"httpx": {
"handlers": ["default"],
"level": "DEBUG",
"level": "INFO",
"propagate": False, # Prevent propagation to root logger
},
"httpcore": {
"handlers": ["default"],
"level": "DEBUG",
"level": "INFO",
"propagate": False, # Prevent propagation to root logger
},
},
@@ -0,0 +1 @@
"""Controllers for utility operations."""
@@ -0,0 +1,102 @@
"""Controller for notes search functionality."""
from typing import Any, Dict, List
class NotesSearchController:
"""Handles notes search logic and scoring."""
def search_notes(
self, notes: List[Dict[str, Any]], query: str
) -> List[Dict[str, Any]]:
"""
Search notes using token-based matching with relevance ranking.
Returns notes sorted by relevance score.
"""
search_results = []
query_tokens = self._process_query(query)
# If empty query after processing, return empty results
if not query_tokens:
return []
# Process and score each note
for note in notes:
title_tokens, content_tokens = self._process_note_content(note)
score = self._calculate_score(query_tokens, title_tokens, content_tokens)
# Only include notes with a non-zero score
if score >= 0.5:
search_results.append(
{
"id": note.get("id"),
"title": note.get("title"),
"category": note.get("category"),
"modified": note.get("modified"),
"_score": score, # Include score for sorting
}
)
# Sort by score in descending order
search_results.sort(key=lambda x: x["_score"], reverse=True)
return search_results
def _process_query(self, query: str) -> List[str]:
"""
Tokenize and normalize the search query.
"""
# Convert to lowercase and split into tokens
tokens = query.lower().split()
# Filter out very short tokens
tokens = [token for token in tokens if len(token) > 1]
return tokens
def _process_note_content(
self, note: Dict[str, Any]
) -> tuple[List[str], List[str]]:
"""
Tokenize and normalize note title and content.
"""
# Process title
title = note.get("title", "").lower()
title_tokens = title.split()
# Process content
content = note.get("content", "").lower()
content_tokens = content.split()
return title_tokens, content_tokens
def _calculate_score(
self,
query_tokens: List[str],
title_tokens: List[str],
content_tokens: List[str],
) -> float:
"""
Calculate a relevance score for a note based on query tokens.
"""
# Constants for weighting
TITLE_WEIGHT = 3.0
CONTENT_WEIGHT = 1.0
score = 0.0
# Count matches in title
title_matches = sum(1 for qt in query_tokens if qt in title_tokens)
if query_tokens: # Avoid division by zero
title_match_ratio = title_matches / len(query_tokens)
score += TITLE_WEIGHT * title_match_ratio
# Count matches in content
content_matches = sum(1 for qt in query_tokens if qt in content_tokens)
if query_tokens: # Avoid division by zero
content_match_ratio = content_matches / len(query_tokens)
score += CONTENT_WEIGHT * content_match_ratio
# If no tokens matched at all, return zero
if title_matches == 0 and content_matches == 0:
return 0.0
return score
-147
View File
@@ -1,147 +0,0 @@
# server.py
import logging
from nextcloud_mcp_server.config import setup_logging
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import FastMCP, Context
from collections.abc import AsyncIterator
from nextcloud_mcp_server.client import NextcloudClient
setup_logging()
@dataclass
class AppContext:
client: NextcloudClient
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
logging.info("Creating Nextcloud client")
client = NextcloudClient.from_env()
logging.info("Client initialization wait complete.")
try:
yield AppContext(client=client)
finally:
# Cleanup on shutdown
await client._client.aclose()
# Create an MCP server
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan)
logger = logging.getLogger(__name__)
@mcp.resource("nc://capabilities")
async def nc_get_capabilities():
"""Get the Nextcloud Host capabilities"""
# client = NextcloudClient.from_env()
ctx = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.capabilities()
@mcp.resource("notes://settings")
async def notes_get_settings():
"""Get the Notes App settings"""
ctx = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_get_settings()
@mcp.tool()
async def nc_get_note(note_id: int, ctx: Context):
"""Get user note using note id"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_get_note(note_id=note_id)
@mcp.tool()
async def nc_notes_create_note(title: str, content: str, category: str, ctx: Context):
"""Create a new note"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_create_note(
title=title,
content=content,
category=category,
)
@mcp.tool()
async def nc_notes_update_note(
note_id: int,
etag: str,
title: str | None,
content: str | None,
category: str | None,
ctx: Context,
):
logger.info("Updating note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_update_note(
note_id=note_id,
etag=etag,
title=title,
content=content,
category=category,
)
@mcp.tool()
async def nc_notes_append_content(note_id: int, content: str, ctx: Context):
"""Append content to an existing note with a clear separator"""
logger.info("Appending content to note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_append_content(note_id=note_id, content=content)
@mcp.tool()
async def nc_notes_search_notes(query: str, ctx: Context):
"""Search notes by title or content, returning only id, title, and category."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_search_notes(query=query)
@mcp.tool()
async def nc_notes_delete_note(note_id: int, ctx: Context):
logger.info("Deleting note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_delete_note(note_id=note_id)
@mcp.resource("nc://Notes/{note_id}/attachments/{attachment_filename}")
async def nc_notes_get_attachment(note_id: int, attachment_filename: str):
"""Get a specific attachment from a note"""
ctx = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Assuming a method get_note_attachment exists in the client
# This method should return the raw content and determine the mime type
content, mime_type = await client.get_note_attachment(
note_id=note_id, filename=attachment_filename
)
return {
"contents": [
{
# Use uppercase 'Notes' to match the decorator
"uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
"mimeType": mime_type, # Client needs to determine this
"data": content, # Return raw bytes/data
}
]
}
def run():
mcp.run()
if __name__ == "__main__":
logger.info("Starting now")
mcp.run()
+13
View File
@@ -0,0 +1,13 @@
from .calendar import configure_calendar_tools
from .notes import configure_notes_tools
from .tables import configure_tables_tools
from .webdav import configure_webdav_tools
from .contacts import configure_contacts_tools
__all__ = [
"configure_calendar_tools",
"configure_notes_tools",
"configure_tables_tools",
"configure_webdav_tools",
"configure_contacts_tools",
]
+794
View File
@@ -0,0 +1,794 @@
import datetime as dt
import logging
from typing import Optional
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_calendar_tools(mcp: FastMCP):
# Calendar tools
@mcp.tool()
async def nc_calendar_list_calendars(ctx: Context):
"""List all available calendars for the user"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.calendar.list_calendars()
@mcp.tool()
async def nc_calendar_create_event(
calendar_name: str,
title: str,
start_datetime: str,
ctx: Context,
end_datetime: str = "",
all_day: bool = False,
description: str = "",
location: str = "",
categories: str = "",
recurring: bool = False,
recurrence_rule: str = "",
recurrence_end_date: str = "",
reminder_minutes: int = 15,
reminder_email: bool = False,
status: str = "CONFIRMED",
priority: int = 5,
privacy: str = "PUBLIC",
attendees: str = "",
url: str = "",
color: str = "",
):
"""Create a comprehensive calendar event with full feature support
Args:
calendar_name: Name of the calendar to create the event in
title: Event title
start_datetime: ISO format: "2025-01-15T14:00:00" or "2025-01-15" for all-day
ctx: MCP context
end_datetime: ISO format end time, empty for all-day events
all_day: Whether this is an all-day event
description: Event description/details
location: Event location
categories: Comma-separated categories (e.g., "work,meeting")
recurring: Whether this is a recurring event
recurrence_rule: RFC5545 RRULE (e.g., "FREQ=WEEKLY;BYDAY=MO,WE,FR")
recurrence_end_date: When to stop recurring
reminder_minutes: Minutes before event to send reminder
reminder_email: Whether to send email notification
status: Event status: CONFIRMED, TENTATIVE, or CANCELLED
priority: Priority level 1-9 (1=highest, 9=lowest, 5=normal)
privacy: Privacy level: PUBLIC, PRIVATE, or CONFIDENTIAL
attendees: Comma-separated email addresses
url: Related URL for the event
color: Event color (hex or name)
Returns:
Dict with event creation result
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
event_data = {
"title": title,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"all_day": all_day,
"description": description,
"location": location,
"categories": categories,
"recurring": recurring,
"recurrence_rule": recurrence_rule,
"recurrence_end_date": recurrence_end_date,
"reminder_minutes": reminder_minutes,
"reminder_email": reminder_email,
"status": status,
"priority": priority,
"privacy": privacy,
"attendees": attendees,
"url": url,
"color": color,
}
return await client.calendar.create_event(calendar_name, event_data)
@mcp.tool()
async def nc_calendar_list_events(
calendar_name: str,
ctx: Context,
start_date: str = "",
end_date: str = "",
limit: int = 50,
min_attendees: Optional[int] = None,
min_duration_minutes: Optional[int] = None,
categories: Optional[str] = None,
status: Optional[str] = None,
title_contains: Optional[str] = None,
location_contains: Optional[str] = None,
search_all_calendars: bool = False,
):
"""List events in a calendar (or all calendars) within date range with advanced filtering.
Args:
calendar_name: Name of the calendar to search. Ignored if search_all_calendars=True.
ctx: MCP context
start_date: Start date for search (YYYY-MM-DD format, e.g., "2025-01-01")
end_date: End date for search (YYYY-MM-DD format, e.g., "2025-01-31")
limit: Maximum number of events to return
min_attendees: Filter events with at least this many attendees
min_duration_minutes: Filter events with at least this duration
categories: Filter events containing any of these categories (comma-separated, e.g., "work,meeting")
status: Filter events by status (CONFIRMED, TENTATIVE, or CANCELLED)
title_contains: Filter events where title contains this text
location_contains: Filter events where location contains this text
search_all_calendars: If True, search across all calendars instead of just one
Returns:
List of events matching the filters
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Convert YYYY-MM-DD format dates to datetime objects
start_datetime = None
end_datetime = None
if start_date:
try:
start_datetime = dt.datetime.strptime(start_date, "%Y-%m-%d")
except ValueError:
# If parsing fails, try to parse as ISO format
try:
start_datetime = dt.datetime.fromisoformat(start_date)
except ValueError:
logger.warning(f"Invalid start_date format: {start_date}")
if end_date:
try:
# For end date, set to end of day (23:59:59)
end_datetime = dt.datetime.strptime(end_date, "%Y-%m-%d").replace(
hour=23, minute=59, second=59
)
except ValueError:
# If parsing fails, try to parse as ISO format
try:
end_datetime = dt.datetime.fromisoformat(end_date)
except ValueError:
logger.warning(f"Invalid end_date format: {end_date}")
# Build filters dictionary
filters = {}
if min_attendees is not None:
filters["min_attendees"] = min_attendees
if min_duration_minutes is not None:
filters["min_duration_minutes"] = min_duration_minutes
if categories is not None:
filters["categories"] = [cat.strip() for cat in categories.split(",")]
if status is not None:
filters["status"] = status
if title_contains is not None:
filters["title_contains"] = title_contains
if location_contains is not None:
filters["location_contains"] = location_contains
if search_all_calendars:
# Search across all calendars with filters
events = await client.calendar.search_events_across_calendars(
start_datetime=start_datetime,
end_datetime=end_datetime,
filters=filters if filters else None,
)
return events[:limit]
else:
# Search in specific calendar
events = await client.calendar.get_calendar_events(
calendar_name=calendar_name,
start_datetime=start_datetime,
end_datetime=end_datetime,
limit=limit,
)
# Apply filters if provided
if filters:
events = client.calendar._apply_event_filters(events, filters)
return events
@mcp.tool()
async def nc_calendar_get_event(
calendar_name: str,
event_uid: str,
ctx: Context,
):
"""Get detailed information about a specific event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
event_data, etag = await client.calendar.get_event(calendar_name, event_uid)
return event_data
@mcp.tool()
async def nc_calendar_update_event(
calendar_name: str,
event_uid: str,
ctx: Context,
# All the same parameters as create_event but optional
title: str | None = None,
start_datetime: str | None = None,
end_datetime: str | None = None,
all_day: bool | None = None,
description: str | None = None,
location: str | None = None,
categories: str | None = None,
# Recurrence updates
recurring: bool | None = None,
recurrence_rule: str | None = None,
# Notification updates
reminder_minutes: int | None = None,
reminder_email: bool | None = None,
# Event property updates
status: str | None = None,
priority: int | None = None,
privacy: str | None = None,
attendees: str | None = None,
url: str | None = None,
color: str | None = None,
etag: str = "",
):
"""Update any aspect of an existing event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Build update data with only non-None values
event_data = {}
if title is not None:
event_data["title"] = title
if start_datetime is not None:
event_data["start_datetime"] = start_datetime
if end_datetime is not None:
event_data["end_datetime"] = end_datetime
if all_day is not None:
event_data["all_day"] = all_day
if description is not None:
event_data["description"] = description
if location is not None:
event_data["location"] = location
if categories is not None:
event_data["categories"] = categories
if recurring is not None:
event_data["recurring"] = recurring
if recurrence_rule is not None:
event_data["recurrence_rule"] = recurrence_rule
if reminder_minutes is not None:
event_data["reminder_minutes"] = reminder_minutes
if reminder_email is not None:
event_data["reminder_email"] = reminder_email
if status is not None:
event_data["status"] = status
if priority is not None:
event_data["priority"] = priority
if privacy is not None:
event_data["privacy"] = privacy
if attendees is not None:
event_data["attendees"] = attendees
if url is not None:
event_data["url"] = url
if color is not None:
event_data["color"] = color
return await client.calendar.update_event(
calendar_name, event_uid, event_data, etag
)
@mcp.tool()
async def nc_calendar_delete_event(
calendar_name: str,
event_uid: str,
ctx: Context,
):
"""Delete a calendar event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.calendar.delete_event(calendar_name, event_uid)
@mcp.tool()
async def nc_calendar_create_meeting(
title: str,
date: str,
time: str,
ctx: Context,
duration_minutes: int = 60,
calendar_name: str = "personal",
attendees: str = "",
location: str = "",
description: str = "",
reminder_minutes: int = 15,
):
"""Quick meeting creation with smart defaults
This is a convenience function for creating events with common meeting defaults.
It automatically:
- Calculates end time based on duration
- Sets status to CONFIRMED
- Adds a reminder
- Uses simpler date/time inputs instead of full ISO format
For full control over all event properties, use nc_calendar_create_event instead.
Args:
title: Meeting title
date: Meeting date (YYYY-MM-DD format, e.g., "2025-01-15")
time: Meeting start time (HH:MM format, e.g., "14:00")
ctx: MCP context
duration_minutes: Meeting duration in minutes (default: 60)
calendar_name: Calendar to create the meeting in (default: "personal")
attendees: Comma-separated email addresses of attendees
location: Meeting location
description: Meeting description/agenda
reminder_minutes: Minutes before meeting to send reminder (default: 15)
Returns:
Dict with meeting creation result
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Combine date and time for start_datetime
start_datetime = f"{date}T{time}:00"
# Calculate end_datetime
start_dt = dt.datetime.fromisoformat(start_datetime)
end_dt = start_dt + dt.timedelta(minutes=duration_minutes)
end_datetime = end_dt.isoformat()
event_data = {
"title": title,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"all_day": False,
"description": description,
"location": location,
"attendees": attendees,
"reminder_minutes": reminder_minutes,
"status": "CONFIRMED",
"priority": 5,
"privacy": "PUBLIC",
}
return await client.calendar.create_event(calendar_name, event_data)
@mcp.tool()
async def nc_calendar_get_upcoming_events(
ctx: Context,
calendar_name: str = "", # Empty = all calendars
days_ahead: int = 7,
limit: int = 10,
):
"""Get upcoming events in next N days"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
now = dt.datetime.now()
end_datetime = now + dt.timedelta(days=days_ahead)
if calendar_name:
# Get events from specific calendar
return await client.calendar.get_calendar_events(
calendar_name=calendar_name,
start_datetime=now,
end_datetime=end_datetime,
limit=limit,
)
else:
# Get events from all calendars
all_calendars = await client.calendar.list_calendars()
all_events = []
for calendar in all_calendars:
try:
events = await client.calendar.get_calendar_events(
calendar_name=calendar["name"],
start_datetime=now,
end_datetime=end_datetime,
limit=limit,
)
# Add calendar info to each event
for event in events:
event["calendar_name"] = calendar["name"]
event["calendar_display_name"] = calendar["display_name"]
all_events.extend(events)
except Exception as e:
logger.warning(
f"Error getting events from calendar {calendar['name']}: {e}"
)
continue
# Sort by start time and limit
all_events.sort(key=lambda x: x.get("start_datetime", ""))
return all_events[:limit]
@mcp.tool()
async def nc_calendar_find_availability(
duration_minutes: int,
ctx: Context,
attendees: str = "", # Comma-separated email list
date_range_start: str = "", # "2025-07-28"
date_range_end: str = "", # "2025-08-04"
business_hours_only: bool = True,
exclude_weekends: bool = True,
preferred_times: str = "", # Comma-separated time ranges like "09:00-12:00,14:00-17:00"
):
"""Find available time slots for scheduling meetings.
This tool intelligently analyzes existing calendar events to find free time slots
that work for all specified attendees within the given constraints.
Args:
duration_minutes: Required duration for the meeting in minutes
attendees: Comma-separated list of attendee email addresses to check availability for
date_range_start: Start date for availability search (YYYY-MM-DD)
date_range_end: End date for availability search (YYYY-MM-DD)
business_hours_only: Only suggest slots during business hours (9 AM - 5 PM)
exclude_weekends: Skip weekends when finding availability
preferred_times: Preferred time ranges as "HH:MM-HH:MM" (comma-separated)
Returns:
List of available time slots with start/end times and duration
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Parse attendees
attendee_list = []
if attendees:
attendee_list = [
email.strip() for email in attendees.split(",") if email.strip()
]
# Parse preferred times
preferred_time_list = []
if preferred_times:
preferred_time_list = [
time_range.strip()
for time_range in preferred_times.split(",")
if time_range.strip()
]
# Convert date strings to datetime objects
start_datetime = None
end_datetime = None
if date_range_start:
try:
start_datetime = dt.datetime.strptime(date_range_start, "%Y-%m-%d")
except ValueError:
logger.warning(f"Invalid date_range_start format: {date_range_start}")
if date_range_end:
try:
end_datetime = dt.datetime.strptime(date_range_end, "%Y-%m-%d").replace(
hour=23, minute=59, second=59
)
except ValueError:
logger.warning(f"Invalid date_range_end format: {date_range_end}")
# Build constraints
constraints = {
"business_hours_only": business_hours_only,
"exclude_weekends": exclude_weekends,
"preferred_times": preferred_time_list,
}
return await client.calendar.find_availability(
duration_minutes=duration_minutes,
attendees=attendee_list,
start_datetime=start_datetime,
end_datetime=end_datetime,
constraints=constraints,
)
@mcp.tool()
async def nc_calendar_bulk_operations(
operation: str, # "update", "delete", "move"
ctx: Context,
title_contains: Optional[str] = None,
categories: Optional[str] = None, # Comma-separated
calendar_name: Optional[str] = None,
start_date: str = "", # "2025-07-01"
end_date: str = "", # "2025-07-31"
status: Optional[str] = None,
location_contains: Optional[str] = None,
# Update operation parameters
new_title: Optional[str] = None,
new_description: Optional[str] = None,
new_location: Optional[str] = None,
new_categories: Optional[str] = None,
new_priority: Optional[int] = None,
new_reminder_minutes: Optional[int] = None,
# Move operation parameters
target_calendar: Optional[str] = None,
):
"""Perform bulk operations (update/delete) on events matching filter criteria.
This tool allows you to efficiently modify or delete multiple events at once
by applying filters to find matching events and then performing the specified operation.
Args:
operation: Type of operation - "update" or "delete"
title_contains: Filter events where title contains this text
categories: Filter events containing any of these categories (comma-separated)
calendar_name: Filter events from this specific calendar
start_date: Filter events starting from this date (YYYY-MM-DD)
end_date: Filter events ending before this date (YYYY-MM-DD)
status: Filter events by status (CONFIRMED, TENTATIVE, CANCELLED)
location_contains: Filter events where location contains this text
# For update operations:
new_title: New title for matching events
new_description: New description for matching events
new_location: New location for matching events
new_categories: New categories for matching events (comma-separated)
new_priority: New priority for matching events (1-9, 5=normal)
new_reminder_minutes: New reminder time in minutes before event
# For move operations:
target_calendar: Calendar to move events to (requires operation="move")
Returns:
Summary of operation results including counts and details
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
if operation not in ["update", "delete", "move"]:
raise ValueError("Operation must be 'update', 'delete', or 'move'")
# Convert date strings to datetime objects
start_datetime = None
end_datetime = None
if start_date:
try:
start_datetime = dt.datetime.strptime(start_date, "%Y-%m-%d")
except ValueError:
logger.warning(f"Invalid start_date format: {start_date}")
if end_date:
try:
end_datetime = dt.datetime.strptime(end_date, "%Y-%m-%d").replace(
hour=23, minute=59, second=59
)
except ValueError:
logger.warning(f"Invalid end_date format: {end_date}")
# Build filter criteria
filter_criteria = {}
if title_contains is not None:
filter_criteria["title_contains"] = title_contains
if categories is not None:
filter_criteria["categories"] = [
cat.strip() for cat in categories.split(",")
]
if status is not None:
filter_criteria["status"] = status
if location_contains is not None:
filter_criteria["location_contains"] = location_contains
# Add datetime strings for client compatibility
if start_date:
filter_criteria["start_date"] = start_date
if end_date:
filter_criteria["end_date"] = end_date
if operation == "delete":
# Find matching events and delete them
if calendar_name:
events = await client.calendar.get_calendar_events(
calendar_name=calendar_name,
start_datetime=start_datetime,
end_datetime=end_datetime,
)
if filter_criteria:
events = client.calendar._apply_event_filters(
events, filter_criteria
)
else:
events = await client.calendar.search_events_across_calendars(
start_datetime=start_datetime,
end_datetime=end_datetime,
filters=filter_criteria,
)
deleted_count = 0
failed_count = 0
results = []
for event in events:
try:
await client.calendar.delete_event(
event.get("calendar_name", calendar_name), event["uid"]
)
deleted_count += 1
results.append(
{
"uid": event["uid"],
"status": "deleted",
"title": event.get("title", ""),
}
)
except Exception as e:
failed_count += 1
results.append(
{
"uid": event["uid"],
"status": "failed",
"error": str(e),
"title": event.get("title", ""),
}
)
return {
"operation": "delete",
"total_found": len(events),
"deleted_count": deleted_count,
"failed_count": failed_count,
"results": results,
}
elif operation == "update":
# Build update data
update_data = {}
if new_title is not None:
update_data["title"] = new_title
if new_description is not None:
update_data["description"] = new_description
if new_location is not None:
update_data["location"] = new_location
if new_categories is not None:
update_data["categories"] = new_categories
if new_priority is not None:
update_data["priority"] = new_priority
if new_reminder_minutes is not None:
update_data["reminder_minutes"] = new_reminder_minutes
if not update_data:
raise ValueError("No update data provided for update operation")
return await client.calendar.bulk_update_events(
filter_criteria, update_data
)
elif operation == "move":
if not target_calendar:
raise ValueError("target_calendar is required for move operation")
# Find matching events
if calendar_name:
events = await client.calendar.get_calendar_events(
calendar_name=calendar_name,
start_datetime=start_datetime,
end_datetime=end_datetime,
)
if filter_criteria:
events = client.calendar._apply_event_filters(
events, filter_criteria
)
else:
events = await client.calendar.search_events_across_calendars(
start_datetime=start_datetime,
end_datetime=end_datetime,
filters=filter_criteria,
)
moved_count = 0
failed_count = 0
results = []
for event in events:
try:
# Create event in target calendar
event_data = {
k: v
for k, v in event.items()
if k
not in [
"uid",
"href",
"etag",
"calendar_name",
"calendar_display_name",
]
}
await client.calendar.create_event(target_calendar, event_data)
# Delete from source calendar
await client.calendar.delete_event(
event.get("calendar_name", calendar_name), event["uid"]
)
moved_count += 1
results.append(
{
"uid": event["uid"],
"status": "moved",
"title": event.get("title", ""),
"from_calendar": event.get("calendar_name", calendar_name),
"to_calendar": target_calendar,
}
)
except Exception as e:
failed_count += 1
results.append(
{
"uid": event["uid"],
"status": "failed",
"error": str(e),
"title": event.get("title", ""),
}
)
return {
"operation": "move",
"total_found": len(events),
"moved_count": moved_count,
"failed_count": failed_count,
"target_calendar": target_calendar,
"results": results,
}
@mcp.tool()
async def nc_calendar_manage_calendar(
action: str, # "create", "delete", "update", "list"
ctx: Context,
calendar_name: str = "",
display_name: str = "",
description: str = "",
color: str = "#1976D2", # Default blue color
):
"""Manage calendar creation, deletion, and properties.
This tool provides comprehensive calendar management functionality including
creating new calendars, deleting existing ones, and updating calendar properties.
Args:
action: Action to perform - "create", "delete", "update", or "list"
calendar_name: Internal name for the calendar (required for create/delete/update)
display_name: Human-readable name for the calendar (used for create/update)
description: Description for the calendar (used for create/update)
color: Hex color code for the calendar (e.g., "#1976D2" for blue)
Returns:
Result of the calendar management operation
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
if action == "list":
return await client.calendar.list_calendars()
elif action == "create":
if not calendar_name:
raise ValueError("calendar_name is required for create action")
return await client.calendar.create_calendar(
calendar_name=calendar_name,
display_name=display_name or calendar_name,
description=description,
color=color,
)
elif action == "delete":
if not calendar_name:
raise ValueError("calendar_name is required for delete action")
return await client.calendar.delete_calendar(calendar_name)
elif action == "update":
if not calendar_name:
raise ValueError("calendar_name is required for update action")
# Note: Calendar property updates require additional CalDAV PROPPATCH implementation
# For now, return an informative message
return {
"status": "not_implemented",
"message": "Calendar property updates require PROPPATCH implementation",
"calendar_name": calendar_name,
"requested_changes": {
"display_name": display_name,
"description": description,
"color": color,
},
}
else:
raise ValueError("Action must be 'create', 'delete', 'update', or 'list'")
+65
View File
@@ -0,0 +1,65 @@
import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_contacts_tools(mcp: FastMCP):
# Contacts tools
@mcp.tool()
async def nc_contacts_list_addressbooks(ctx: Context):
"""List all addressbooks for the user."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.list_addressbooks()
@mcp.tool()
async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str):
"""List all addressbooks for the user."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.list_contacts(addressbook=addressbook)
@mcp.tool()
async def nc_contacts_create_addressbook(
ctx: Context, *, name: str, display_name: str
):
"""Create a new addressbook.
Args:
name: The name of the addressbook.
display_name: The display name of the addressbook.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.create_addressbook(
name=name, display_name=display_name
)
@mcp.tool()
async def nc_contacts_delete_addressbook(ctx: Context, *, name: str):
"""Delete an addressbook."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.delete_addressbook(name=name)
@mcp.tool()
async def nc_contacts_create_contact(
ctx: Context, *, addressbook: str, uid: str, contact_data: dict
):
"""Create a new contact.
Args:
addressbook: The name of the addressbook to create the contact in.
uid: The unique ID for the contact.
contact_data: A dictionary with the contact's details, e.g. {"fn": "John Doe", "email": "john.doe@example.com"}.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.create_contact(
addressbook=addressbook, uid=uid, contact_data=contact_data
)
@mcp.tool()
async def nc_contacts_delete_contact(ctx: Context, *, addressbook: str, uid: str):
"""Delete a contact."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.delete_contact(addressbook=addressbook, uid=uid)
+95
View File
@@ -0,0 +1,95 @@
import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_notes_tools(mcp: FastMCP):
@mcp.resource("notes://settings")
async def notes_get_settings():
"""Get the Notes App settings"""
ctx: Context = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.get_settings()
@mcp.resource("nc://Notes/{note_id}/attachments/{attachment_filename}")
async def nc_notes_get_attachment(note_id: int, attachment_filename: str):
"""Get a specific attachment from a note"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Assuming a method get_note_attachment exists in the client
# This method should return the raw content and determine the mime type
content, mime_type = await client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename
)
return {
"contents": [
{
# Use uppercase 'Notes' to match the decorator
"uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
"mimeType": mime_type, # Client needs to determine this
"data": content, # Return raw bytes/data
}
]
}
@mcp.tool()
async def nc_get_note(note_id: int, ctx: Context):
"""Get user note using note id"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.get_note(note_id)
@mcp.tool()
async def nc_notes_create_note(
title: str, content: str, category: str, ctx: Context
):
"""Create a new note"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.create_note(
title=title,
content=content,
category=category,
)
@mcp.tool()
async def nc_notes_update_note(
note_id: int,
etag: str,
title: str | None,
content: str | None,
category: str | None,
ctx: Context,
):
logger.info("Updating note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.update(
note_id=note_id,
etag=etag,
title=title,
content=content,
category=category,
)
@mcp.tool()
async def nc_notes_append_content(note_id: int, content: str, ctx: Context):
"""Append content to an existing note with a clear separator"""
logger.info("Appending content to note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.append_content(note_id=note_id, content=content)
@mcp.tool()
async def nc_notes_search_notes(query: str, ctx: Context):
"""Search notes by title or content, returning only id, title, and category."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes_search_notes(query=query)
@mcp.tool()
async def nc_notes_delete_note(note_id: int, ctx: Context):
logger.info("Deleting note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.notes.delete_note(note_id)
+57
View File
@@ -0,0 +1,57 @@
import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_tables_tools(mcp: FastMCP):
# Tables tools
@mcp.tool()
async def nc_tables_list_tables(ctx: Context):
"""List all tables available to the user"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.list_tables()
@mcp.tool()
async def nc_tables_get_schema(table_id: int, ctx: Context):
"""Get the schema/structure of a specific table including columns and views"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.get_table_schema(table_id)
@mcp.tool()
async def nc_tables_read_table(
table_id: int,
ctx: Context,
limit: int | None = None,
offset: int | None = None,
):
"""Read rows from a table with optional pagination"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.get_table_rows(table_id, limit, offset)
@mcp.tool()
async def nc_tables_insert_row(table_id: int, data: dict, ctx: Context):
"""Insert a new row into a table.
Data should be a dictionary mapping column IDs to values, e.g. {1: "text", 2: 42}
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.create_row(table_id, data)
@mcp.tool()
async def nc_tables_update_row(row_id: int, data: dict, ctx: Context):
"""Update an existing row in a table.
Data should be a dictionary mapping column IDs to new values, e.g. {1: "new text", 2: 99}
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.update_row(row_id, data)
@mcp.tool()
async def nc_tables_delete_row(row_id: int, ctx: Context):
"""Delete a row from a table"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.tables.delete_row(row_id)
+151
View File
@@ -0,0 +1,151 @@
import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_webdav_tools(mcp: FastMCP):
# WebDAV file system tools
@mcp.tool()
async def nc_webdav_list_directory(ctx: Context, path: str = ""):
"""List files and directories in the specified NextCloud path.
Args:
path: Directory path to list (empty string for root directory)
Returns:
List of items with metadata including name, path, is_directory, size, content_type, last_modified
Examples:
# List root directory
await nc_webdav_list_directory("")
# List a specific folder
await nc_webdav_list_directory("Documents/Projects")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.webdav.list_directory(path)
@mcp.tool()
async def nc_webdav_read_file(path: str, ctx: Context):
"""Read the content of a file from NextCloud.
Args:
path: Full path to the file to read
Returns:
Dict with path, content, content_type, size, and encoding (if binary)
Text files are decoded to UTF-8, binary files are base64 encoded
Examples:
# Read a text file
result = await nc_webdav_read_file("Documents/readme.txt")
print(result['content']) # Decoded text content
# Read a binary file
result = await nc_webdav_read_file("Images/photo.jpg")
print(result['encoding']) # 'base64'
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
content, content_type = await client.webdav.read_file(path)
# For text files, decode content for easier viewing
if content_type and content_type.startswith("text/"):
try:
decoded_content = content.decode("utf-8")
return {
"path": path,
"content": decoded_content,
"content_type": content_type,
"size": len(content),
}
except UnicodeDecodeError:
pass
# For binary files, return metadata and base64 encoded content
import base64
return {
"path": path,
"content": base64.b64encode(content).decode("ascii"),
"content_type": content_type,
"size": len(content),
"encoding": "base64",
}
@mcp.tool()
async def nc_webdav_write_file(
path: str, content: str, ctx: Context, content_type: str | None = None
):
"""Write content to a file in NextCloud.
Args:
path: Full path where to write the file
content: File content (text or base64 for binary)
content_type: MIME type (auto-detected if not provided, use 'type;base64' for binary)
Returns:
Dict with status_code indicating success
Examples:
# Write a text file
await nc_webdav_write_file("Documents/notes.md", "# My Notes\nContent here...")
# Write binary data (base64 encoded)
await nc_webdav_write_file("files/data.bin", base64_content, "application/octet-stream;base64")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
# Handle base64 encoded content
if content_type and "base64" in content_type.lower():
import base64
content_bytes = base64.b64decode(content)
content_type = content_type.replace(";base64", "")
else:
content_bytes = content.encode("utf-8")
return await client.webdav.write_file(path, content_bytes, content_type)
@mcp.tool()
async def nc_webdav_create_directory(path: str, ctx: Context):
"""Create a directory in NextCloud.
Args:
path: Full path of the directory to create
Returns:
Dict with status_code (201 for created, 405 if already exists)
Examples:
# Create a single directory
await nc_webdav_create_directory("NewProject")
# Create nested directories (parent must exist)
await nc_webdav_create_directory("Projects/MyApp/docs")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.webdav.create_directory(path)
@mcp.tool()
async def nc_webdav_delete_resource(path: str, ctx: Context):
"""Delete a file or directory in NextCloud.
Args:
path: Full path of the file or directory to delete
Returns:
Dict with status_code indicating result (404 if not found)
Examples:
# Delete a file
await nc_webdav_delete_resource("old_document.txt")
# Delete a directory (will delete all contents)
await nc_webdav_delete_resource("temp_folder")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.webdav.delete_resource(path)
+7 -8
View File
@@ -1,6 +1,6 @@
[project]
name = "nextcloud-mcp-server"
version = "0.2.5"
version = "0.6.1"
description = ""
authors = [
{name = "Chris Coutinho",email = "chris@coutinho.io"}
@@ -8,21 +8,20 @@ authors = [
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"mcp[cli] (>=1.9,<1.10)",
"mcp[cli] (>=1.10,<1.11)",
"httpx (>=0.28.1,<0.29.0)",
"pillow (>=11.2.1,<12.0.0)"
"pillow (>=11.2.1,<12.0.0)",
"icalendar (>=6.0.0,<7.0.0)",
"pythonvcard4>=0.2.0",
]
[project.scripts]
nc-mcp-server = "nextcloud_mcp_server.server:run"
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_test_loop_scope = "session"
asyncio_default_fixture_loop_scope = "session"
log_cli = 1
log_cli_level = "WARN"
log_level = "WARN"
log_cli_level = "INFO"
log_level = "INFO"
markers = [
"integration: marks tests as slow (deselect with '-m \"not slow\"')"
]
+2 -1
View File
@@ -1,7 +1,8 @@
{
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
"extends": [
"config:best-practices"
"config:best-practices",
"mergeConfidence:all-badges"
],
"dependencyDashboard": true
}
+139 -14
View File
@@ -1,17 +1,20 @@
import pytest
import os
import logging
import os
import uuid
from nextcloud_mcp_server.client import NextcloudClient, HTTPStatusError
import asyncio
from typing import Any, AsyncGenerator
import pytest
from httpx import HTTPStatusError
from mcp import ClientSession
from mcp.client.sse import sse_client
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# pytestmark = pytest.mark.asyncio(loop_scope="package")
@pytest.fixture(scope="session")
async def nc_client() -> NextcloudClient:
@pytest.fixture(scope="module")
async def nc_client() -> AsyncGenerator[NextcloudClient, Any]:
"""
Fixture to create a NextcloudClient instance for integration tests.
Uses environment variables for configuration.
@@ -28,10 +31,54 @@ async def nc_client() -> NextcloudClient:
logger.info(
"NextcloudClient session fixture initialized and capabilities checked."
)
yield client
except Exception as e:
logger.error(f"Failed to initialize NextcloudClient session fixture: {e}")
pytest.fail(f"Failed to connect to Nextcloud or get capabilities: {e}")
return client
finally:
await client.close()
@pytest.fixture
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for integration tests.
"""
logger.info("Creating SSE client")
sse_context = sse_client(url="http://127.0.0.1:8000/sse")
session_context = None
try:
read, write = await sse_context.__aenter__()
session_context = ClientSession(read, write)
session = await session_context.__aenter__()
await session.initialize()
logger.info("MCP client session initialized successfully")
yield session
finally:
# Clean up in reverse order, ignoring task scope issues
if session_context is not None:
try:
await session_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing session: {e}")
except Exception as e:
logger.warning(f"Error closing session: {e}")
try:
await sse_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing SSE client: {e}")
except Exception as e:
logger.warning(f"Error closing SSE client: {e}")
@pytest.fixture
@@ -40,7 +87,6 @@ async def temporary_note(nc_client: NextcloudClient):
Fixture to create a temporary note for a test and ensure its deletion afterward.
Yields the created note dictionary.
"""
asyncio.new_event_loop()
note_id = None
unique_suffix = uuid.uuid4().hex[:8]
@@ -51,7 +97,7 @@ async def temporary_note(nc_client: NextcloudClient):
logger.info(f"Creating temporary note: {note_title}")
try:
created_note_data = await nc_client.notes_create_note(
created_note_data = await nc_client.notes.create_note(
title=note_title, content=note_content, category=note_category
)
note_id = created_note_data.get("id")
@@ -65,7 +111,7 @@ async def temporary_note(nc_client: NextcloudClient):
if note_id:
logger.info(f"Cleaning up temporary note ID: {note_id}")
try:
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Successfully deleted temporary note ID: {note_id}")
except HTTPStatusError as e:
# Ignore 404 if note was already deleted by the test itself
@@ -86,7 +132,6 @@ async def temporary_note_with_attachment(
Yields a tuple: (note_data, attachment_filename, attachment_content).
Depends on the temporary_note fixture.
"""
asyncio.new_event_loop()
note_data = temporary_note
note_id = note_data["id"]
@@ -101,7 +146,7 @@ async def temporary_note_with_attachment(
)
try:
# Pass the category to add_note_attachment
upload_response = await nc_client.add_note_attachment(
upload_response = await nc_client.webdav.add_note_attachment(
note_id=note_id,
filename=attachment_filename,
content=attachment_content,
@@ -125,3 +170,83 @@ async def temporary_note_with_attachment(
# Note: The temporary_note fixture's finally block will handle note deletion,
# which should also trigger the WebDAV directory deletion attempt.
@pytest.fixture(scope="module")
async def temporary_addressbook(nc_client: NextcloudClient):
"""
Fixture to create a temporary addressbook for a test and ensure its deletion afterward.
Yields the created addressbook dictionary.
"""
addressbook_name = f"test-addressbook-{uuid.uuid4().hex[:8]}"
logger.info(f"Creating temporary addressbook: {addressbook_name}")
try:
await nc_client.contacts.create_addressbook(
name=addressbook_name, display_name=f"Test Addressbook {addressbook_name}"
)
logger.info(f"Temporary addressbook created: {addressbook_name}")
yield addressbook_name
finally:
logger.info(f"Cleaning up temporary addressbook: {addressbook_name}")
try:
await nc_client.contacts.delete_addressbook(name=addressbook_name)
logger.info(
f"Successfully deleted temporary addressbook: {addressbook_name}"
)
except HTTPStatusError as e:
if e.response.status_code != 404:
logger.error(
f"HTTP error deleting temporary addressbook {addressbook_name}: {e}"
)
else:
logger.warning(
f"Temporary addressbook {addressbook_name} already deleted (404)."
)
except Exception as e:
logger.error(
f"Unexpected error deleting temporary addressbook {addressbook_name}: {e}"
)
@pytest.fixture
async def temporary_contact(nc_client: NextcloudClient, temporary_addressbook: str):
"""
Fixture to create a temporary contact in a temporary addressbook and ensure its deletion.
Yields the created contact's UID.
"""
contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
addressbook_name = temporary_addressbook
contact_data = {
"fn": "John Doe",
"email": "john.doe@example.com",
"tel": "1234567890",
}
logger.info(f"Creating temporary contact in addressbook: {addressbook_name}")
try:
await nc_client.contacts.create_contact(
addressbook=addressbook_name,
uid=contact_uid,
contact_data=contact_data,
)
logger.info(f"Temporary contact created with UID: {contact_uid}")
yield contact_uid
finally:
logger.info(f"Cleaning up temporary contact: {contact_uid}")
try:
await nc_client.contacts.delete_contact(
addressbook=addressbook_name, uid=contact_uid
)
logger.info(f"Successfully deleted temporary contact: {contact_uid}")
except HTTPStatusError as e:
if e.response.status_code != 404:
logger.error(
f"HTTP error deleting temporary contact {contact_uid}: {e}"
)
else:
logger.warning(
f"Temporary contact {contact_uid} already deleted (404)."
)
except Exception as e:
logger.error(
f"Unexpected error deleting temporary contact {contact_uid}: {e}"
)
+17 -16
View File
@@ -1,7 +1,8 @@
import pytest
import logging
import time
import uuid
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
@@ -29,7 +30,7 @@ async def test_attachments_add_and_get(
f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}"
)
# Pass category to get_note_attachment
retrieved_content, retrieved_mime = await nc_client.get_note_attachment(
retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=note_category
)
logger.info(
@@ -67,7 +68,7 @@ async def test_attachments_add_to_note_with_category(
f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}"
)
# Pass category to add_note_attachment
upload_response = await nc_client.add_note_attachment(
upload_response = await nc_client.webdav.add_note_attachment(
note_id=note_id,
filename=attachment_filename,
content=attachment_content,
@@ -86,7 +87,7 @@ async def test_attachments_add_to_note_with_category(
f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}"
)
# Pass category to get_note_attachment
retrieved_content, retrieved_mime = await nc_client.get_note_attachment(
retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
note_id=note_id,
filename=attachment_filename,
category=note_category, # Pass the note's category
@@ -127,13 +128,13 @@ async def test_attachments_cleanup_on_note_delete(
# Manually delete the note
logger.info(f"Manually deleting note ID: {note_id} within the test.")
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Note ID: {note_id} deleted successfully.")
time.sleep(1)
# Verify Note Is Deleted
with pytest.raises(HTTPStatusError) as excinfo_note:
await nc_client.notes_get_note(note_id=note_id)
await nc_client.notes.get_note(note_id=note_id)
assert excinfo_note.value.response.status_code == 404
logger.info(f"Verified note {note_id} deletion (404 received).")
@@ -145,7 +146,7 @@ async def test_attachments_cleanup_on_note_delete(
# Pass category to get_note_attachment - although it should fail anyway
# because the note (and thus details) are gone.
# The client method will raise 404 from the initial notes_get_note call.
await nc_client.get_note_attachment(
await nc_client.webdav.get_note_attachment(
note_id=note_id,
filename=attachment_filename,
category=note_category, # Pass category, though note fetch should fail first
@@ -205,7 +206,7 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
try:
# 1. Create note with initial category
logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
created_note = await nc_client.notes_create_note(
created_note = await nc_client.notes.create_note(
title=note_title, content="Initial content", category=initial_category
)
note_id = created_note["id"]
@@ -217,7 +218,7 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info(
f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
)
upload_response = await nc_client.add_note_attachment(
upload_response = await nc_client.webdav.add_note_attachment(
note_id=note_id,
filename=attachment_filename,
content=attachment_content,
@@ -232,7 +233,7 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info(
f"Verifying attachment retrieval from initial category '{initial_category}'"
)
retrieved_content1, _ = await nc_client.get_note_attachment(
retrieved_content1, _ = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category
)
assert retrieved_content1 == attachment_content
@@ -243,9 +244,9 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
)
# Need to fetch the latest etag after attachment add (WebDAV ops don't update note etag)
current_note_data = await nc_client.notes_get_note(note_id=note_id)
current_note_data = await nc_client.notes.get_note(note_id=note_id)
current_etag = current_note_data["etag"]
updated_note = await nc_client.notes_update_note(
updated_note = await nc_client.notes.update(
note_id=note_id,
etag=current_etag,
category=new_category,
@@ -261,7 +262,7 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info(
f"Verifying attachment retrieval from new category '{new_category}'"
)
retrieved_content2, _ = await nc_client.get_note_attachment(
retrieved_content2, _ = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=new_category
)
assert retrieved_content2 == attachment_content
@@ -326,18 +327,18 @@ async def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"Cleaning up note ID: {note_id} (last known category: '{new_category}')"
)
try:
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Note {note_id} deleted.")
time.sleep(1)
# Verify note deletion
with pytest.raises(HTTPStatusError) as excinfo_note_del:
await nc_client.notes_get_note(note_id=note_id)
await nc_client.notes.get_note(note_id=note_id)
assert excinfo_note_del.value.response.status_code == 404
logger.info("Verified note deleted (404).")
# Verify attachment deletion (should fail with 404 on the initial note fetch)
with pytest.raises(HTTPStatusError) as excinfo_attach_del:
# Pass the *last known* category, although the note fetch should fail first
await nc_client.get_note_attachment(
await nc_client.webdav.get_note_attachment(
note_id=note_id,
filename=attachment_filename,
category=new_category,
@@ -0,0 +1,426 @@
"""Integration tests for Calendar CalDAV operations."""
import logging
import uuid
from datetime import datetime, timedelta
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration
@pytest.fixture
def test_calendar_name():
"""Unique calendar name for testing."""
return f"test_calendar_{uuid.uuid4().hex[:8]}"
@pytest.fixture
async def temporary_calendar(nc_client: NextcloudClient, test_calendar_name: str):
"""Create a temporary calendar for testing and clean up afterward."""
calendar_name = test_calendar_name
try:
# Create a test calendar
logger.info(f"Creating temporary calendar: {calendar_name}")
result = await nc_client.calendar.create_calendar(
calendar_name=calendar_name,
display_name=f"Test Calendar {calendar_name}",
description="Temporary calendar for integration testing",
color="#FF5722",
)
if result["status_code"] not in [200, 201]:
pytest.skip(f"Failed to create temporary calendar: {result}")
logger.info(f"Created temporary calendar: {calendar_name}")
yield calendar_name
except Exception as e:
logger.error(f"Error setting up temporary calendar: {e}")
pytest.skip(f"Calendar setup failed: {e}")
finally:
# Cleanup: Delete the temporary calendar
try:
logger.info(f"Cleaning up temporary calendar: {calendar_name}")
await nc_client.calendar.delete_calendar(calendar_name)
logger.info(f"Successfully deleted temporary calendar: {calendar_name}")
except Exception as e:
logger.error(f"Error deleting temporary calendar {calendar_name}: {e}")
@pytest.fixture
async def temporary_event(nc_client: NextcloudClient, temporary_calendar: str):
"""Create a temporary event for testing and clean up afterward."""
event_uid = None
calendar_name = temporary_calendar
# Create a test event
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": f"Test Event {uuid.uuid4().hex[:8]}",
"start_datetime": tomorrow.strftime("%Y-%m-%dT14:00:00"),
"end_datetime": tomorrow.strftime("%Y-%m-%dT15:00:00"),
"description": "Test event created by integration tests",
"location": "Test Location",
"categories": "testing",
"status": "CONFIRMED",
"priority": 5,
}
try:
logger.info(f"Creating temporary event in calendar: {calendar_name}")
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result.get("uid")
if not event_uid:
pytest.fail("Failed to create temporary event")
logger.info(f"Created temporary event with UID: {event_uid}")
yield {"uid": event_uid, "calendar_name": calendar_name, "data": event_data}
finally:
# Cleanup
if event_uid:
try:
logger.info(f"Cleaning up temporary event: {event_uid}")
await nc_client.calendar.delete_event(calendar_name, event_uid)
logger.info(f"Successfully deleted temporary event: {event_uid}")
except HTTPStatusError as e:
if e.response.status_code != 404:
logger.error(f"Error deleting temporary event {event_uid}: {e}")
except Exception as e:
logger.error(
f"Unexpected error deleting temporary event {event_uid}: {e}"
)
async def test_list_calendars(nc_client: NextcloudClient):
"""Test listing available calendars."""
calendars = await nc_client.calendar.list_calendars()
assert isinstance(calendars, list)
if not calendars:
pytest.skip("No calendars available - Calendar app may not be enabled")
logger.info(f"Found {len(calendars)} calendars")
# Check structure of calendars
for calendar in calendars:
assert "name" in calendar
assert "display_name" in calendar
assert "href" in calendar
# Optional fields
assert "description" in calendar
assert "color" in calendar
logger.info(f"Calendar: {calendar['name']} - {calendar['display_name']}")
async def test_create_and_delete_event(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating and deleting a basic event."""
calendar_name = temporary_calendar
# Create event
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": "Integration Test Event",
"start_datetime": tomorrow.strftime("%Y-%m-%dT10:00:00"),
"end_datetime": tomorrow.strftime("%Y-%m-%dT11:00:00"),
"description": "Test event for integration testing",
"location": "Test Room",
"categories": "testing,integration",
"status": "CONFIRMED",
"priority": 3,
}
try:
result = await nc_client.calendar.create_event(calendar_name, event_data)
assert "uid" in result
assert result["status_code"] in [200, 201, 204]
event_uid = result["uid"]
logger.info(f"Created event with UID: {event_uid}")
# Verify event was created by retrieving it
retrieved_event, etag = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["uid"] == event_uid
assert retrieved_event["title"] == "Integration Test Event"
assert retrieved_event["location"] == "Test Room"
# Delete event
delete_result = await nc_client.calendar.delete_event(calendar_name, event_uid)
assert delete_result["status_code"] in [200, 204, 404]
logger.info(f"Successfully deleted event: {event_uid}")
except Exception as e:
logger.error(f"Test failed: {e}")
raise
async def test_create_all_day_event(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating an all-day event."""
calendar_name = temporary_calendar
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": "All Day Test Event",
"start_datetime": tomorrow.strftime("%Y-%m-%d"),
"all_day": True,
"description": "Test all-day event",
"categories": "testing",
}
try:
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created all-day event with UID: {event_uid}")
# Verify event
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "All Day Test Event"
assert retrieved_event.get("all_day") is True
# Cleanup
await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e:
logger.error(f"All-day event test failed: {e}")
raise
async def test_create_recurring_event(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating a recurring event."""
calendar_name = temporary_calendar
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": "Weekly Recurring Test",
"start_datetime": tomorrow.strftime("%Y-%m-%dT14:00:00"),
"end_datetime": tomorrow.strftime("%Y-%m-%dT15:00:00"),
"description": "Test recurring event",
"recurring": True,
"recurrence_rule": "FREQ=WEEKLY;BYDAY=MO,WE,FR",
"reminder_minutes": 30,
}
try:
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created recurring event with UID: {event_uid}")
# Verify event
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "Weekly Recurring Test"
assert retrieved_event.get("recurring") is True
# Cleanup
await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e:
logger.error(f"Recurring event test failed: {e}")
raise
async def test_list_events_in_range(nc_client: NextcloudClient, temporary_event: dict):
"""Test listing events within a date range."""
calendar_name = temporary_event["calendar_name"]
# Get events for the next week
start_datetime = datetime.now()
end_datetime = datetime.now() + timedelta(days=7)
events = await nc_client.calendar.get_calendar_events(
calendar_name=calendar_name,
start_datetime=start_datetime,
end_datetime=end_datetime,
limit=50,
)
assert isinstance(events, list)
logger.info(f"Found {len(events)} events in date range")
# Our temporary event should be in the list
event_uids = [event.get("uid") for event in events]
assert temporary_event["uid"] in event_uids
# Check event structure
for event in events:
assert "uid" in event
assert "title" in event
assert "start_datetime" in event
async def test_update_event(nc_client: NextcloudClient, temporary_event: dict):
"""Test updating an existing event."""
calendar_name = temporary_event["calendar_name"]
event_uid = temporary_event["uid"]
# Update event data
updated_data = {
"title": "Updated Test Event Title",
"description": "Updated description for test event",
"location": "Updated Location",
"priority": 1, # High priority
}
try:
result = await nc_client.calendar.update_event(
calendar_name, event_uid, updated_data
)
assert result["uid"] == event_uid
# Verify updates
updated_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid)
assert updated_event["title"] == "Updated Test Event Title"
assert updated_event["description"] == "Updated description for test event"
assert updated_event["location"] == "Updated Location"
assert updated_event["priority"] == 1
logger.info(f"Successfully updated event: {event_uid}")
except Exception as e:
logger.error(f"Event update test failed: {e}")
raise
async def test_create_event_with_attendees(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating an event with attendees."""
calendar_name = temporary_calendar
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": "Meeting with Attendees",
"start_datetime": tomorrow.strftime("%Y-%m-%dT16:00:00"),
"end_datetime": tomorrow.strftime("%Y-%m-%dT17:00:00"),
"description": "Test meeting with multiple attendees",
"location": "Conference Room A",
"attendees": "test1@example.com,test2@example.com",
"reminder_minutes": 15,
"status": "TENTATIVE",
}
try:
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created event with attendees, UID: {event_uid}")
# Verify event
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "Meeting with Attendees"
assert "test1@example.com" in retrieved_event.get("attendees", "")
assert retrieved_event["status"] == "TENTATIVE"
# Cleanup
await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e:
logger.error(f"Event with attendees test failed: {e}")
raise
async def test_get_nonexistent_event(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test retrieving a non-existent event."""
calendar_name = temporary_calendar
fake_uid = f"nonexistent-{uuid.uuid4()}"
with pytest.raises(HTTPStatusError) as exc_info:
await nc_client.calendar.get_event(calendar_name, fake_uid)
assert exc_info.value.response.status_code == 404
logger.info(f"Correctly got 404 for nonexistent event: {fake_uid}")
async def test_delete_nonexistent_event(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test deleting a non-existent event."""
calendar_name = temporary_calendar
fake_uid = f"nonexistent-{uuid.uuid4()}"
result = await nc_client.calendar.delete_event(calendar_name, fake_uid)
assert result["status_code"] == 404
logger.info(f"Correctly got 404 for deleting nonexistent event: {fake_uid}")
async def test_event_with_url_and_categories(
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating an event with URL and multiple categories."""
calendar_name = temporary_calendar
tomorrow = datetime.now() + timedelta(days=1)
event_data = {
"title": "Event with URL and Categories",
"start_datetime": tomorrow.strftime("%Y-%m-%dT09:00:00"),
"end_datetime": tomorrow.strftime("%Y-%m-%dT10:30:00"),
"description": "Test event with additional metadata",
"categories": "work,meeting,important,quarterly",
"url": "https://zoom.us/j/123456789",
"privacy": "PRIVATE",
"priority": 2,
}
try:
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created event with metadata, UID: {event_uid}")
# Verify event
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "Event with URL and Categories"
assert "work" in retrieved_event.get("categories", "")
assert "important" in retrieved_event.get("categories", "")
assert retrieved_event.get("url") == "https://zoom.us/j/123456789"
assert retrieved_event.get("privacy") == "PRIVATE"
assert retrieved_event.get("priority") == 2
# Cleanup
await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e:
logger.error(f"Event with metadata test failed: {e}")
raise
async def test_calendar_operations_error_handling(
nc_client: NextcloudClient,
):
"""Test error handling for calendar operations."""
# Test with non-existent calendar
fake_calendar = f"nonexistent_calendar_{uuid.uuid4().hex}"
with pytest.raises(HTTPStatusError):
await nc_client.calendar.get_calendar_events(fake_calendar)
logger.info("Error handling tests completed successfully")
+86
View File
@@ -0,0 +1,86 @@
"""Integration tests for Contacts MCP tools."""
import logging
import uuid
import pytest
from mcp import ClientSession
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
async def test_mcp_contacts_workflow(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test complete Contacts workflow via MCP tools with verification via NextcloudClient."""
addressbook_name = f"mcp-test-addressbook-{uuid.uuid4().hex[:8]}"
unique_suffix = uuid.uuid4().hex[:8]
contact_uid = f"mcp-contact-{unique_suffix}"
contact_data = {
"fn": f"MCP Contact {unique_suffix}",
"email": f"mcp.contact.{unique_suffix}@example.com",
"tel": "1234567890",
}
try:
# 1. Create address book via MCP
logger.info(f"Creating address book via MCP: {addressbook_name}")
create_ab_result = await nc_mcp_client.call_tool(
"nc_contacts_create_addressbook",
{"name": addressbook_name, "display_name": f"MCP Test {addressbook_name}"},
)
assert create_ab_result.isError is False
# 2. Verify address book creation
addressbooks = await nc_client.contacts.list_addressbooks()
assert any(ab["name"] == addressbook_name for ab in addressbooks)
# 3. Create contact via MCP
logger.info(f"Creating contact in {addressbook_name} via MCP")
create_c_result = await nc_mcp_client.call_tool(
"nc_contacts_create_contact",
{
"addressbook": addressbook_name,
"uid": contact_uid,
"contact_data": contact_data,
},
)
assert create_c_result.isError is False
# 4. Verify contact creation
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
assert any(c["vcard_id"] == contact_uid for c in contacts)
# 5. Delete contact via MCP
logger.info(f"Deleting contact {contact_uid} via MCP")
delete_c_result = await nc_mcp_client.call_tool(
"nc_contacts_delete_contact",
{"addressbook": addressbook_name, "uid": contact_uid},
)
assert delete_c_result.isError is False
# 6. Verify contact deletion
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
assert not any(c["vcard_id"] == contact_uid for c in contacts)
# 7. Delete address book via MCP
logger.info(f"Deleting address book {addressbook_name} via MCP")
delete_ab_result = await nc_mcp_client.call_tool(
"nc_contacts_delete_addressbook", {"name": addressbook_name}
)
assert delete_ab_result.isError is False
# 8. Verify address book deletion
addressbooks = await nc_client.contacts.list_addressbooks()
assert not any(ab["name"] == addressbook_name for ab in addressbooks)
finally:
# Cleanup in case of failure
try:
await nc_client.contacts.delete_addressbook(name=addressbook_name)
except Exception:
pass
@@ -0,0 +1,88 @@
"""Integration tests for Contacts CardDAV operations."""
import logging
import uuid
import pytest
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration
async def test_list_addressbooks(nc_client: NextcloudClient):
"""Test listing available addressbooks."""
addressbooks = await nc_client.contacts.list_addressbooks()
assert isinstance(addressbooks, list)
if not addressbooks:
pytest.skip("No addressbooks available - Contacts app may not be enabled")
logger.info(f"Found {len(addressbooks)} addressbooks")
# Check structure of addressbooks
for addressbook in addressbooks:
assert "name" in addressbook
assert "display_name" in addressbook
assert "getctag" in addressbook
logger.info(
f"Addressbook: {addressbook['name']} - {addressbook['display_name']}"
)
async def test_create_and_delete_addressbook(
nc_client: NextcloudClient, temporary_addressbook: str
):
"""Test creating and deleting a basic addressbook."""
addressbooks = await nc_client.contacts.list_addressbooks()
addressbook_names = [ab["name"] for ab in addressbooks]
assert temporary_addressbook in addressbook_names
async def test_list_contacts(
nc_client: NextcloudClient, temporary_addressbook: str, temporary_contact: str
):
"""Test listing contacts in an addressbook."""
contacts = await nc_client.contacts.list_contacts(addressbook=temporary_addressbook)
contact_uids = [c["vcard_id"] for c in contacts]
assert temporary_contact in contact_uids
async def test_full_contact_workflow(
nc_client: NextcloudClient, temporary_addressbook: str
):
"""Test the full workflow of creating, retrieving, and deleting a contact."""
addressbook_name = temporary_addressbook
contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
contact_data = {
"fn": "Jane Doe",
"email": "jane.doe@example.com",
"tel": "9876543210",
}
# Create contact
await nc_client.contacts.create_contact(
addressbook=addressbook_name,
uid=contact_uid,
contact_data=contact_data,
)
# Verify contact was created by listing
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
contact_uids = [c["vcard_id"] for c in contacts]
assert contact_uid in contact_uids
# Delete contact
await nc_client.contacts.delete_contact(
addressbook=addressbook_name, uid=contact_uid
)
# Verify contact was deleted
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
contact_uids = [c["vcard_id"] for c in contacts]
assert contact_uid not in contact_uids
+10 -9
View File
@@ -1,10 +1,11 @@
import pytest
import logging
import time
import uuid
import logging
from PIL import Image, ImageDraw
from io import BytesIO
import pytest
from httpx import HTTPStatusError # Import if needed for specific error checks
from PIL import Image, ImageDraw
from nextcloud_mcp_server.client import NextcloudClient
@@ -62,7 +63,7 @@ async def test_note_with_embedded_image(
logger.info(
f"Uploading image attachment '{attachment_filename}' to note {note_id} (category: '{note_category or ''}')..."
)
upload_response = await nc_client.add_note_attachment(
upload_response = await nc_client.webdav.add_note_attachment(
note_id=note_id,
filename=attachment_filename,
content=image_content,
@@ -115,7 +116,7 @@ async def test_note_with_embedded_image(
<img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="150" />
"""
logger.info("Updating note content with image references...")
updated_note = await nc_client.notes_update_note(
updated_note = await nc_client.notes.update(
note_id=note_id,
etag=note_etag, # Use etag from the created note
content=updated_content,
@@ -128,7 +129,7 @@ async def test_note_with_embedded_image(
time.sleep(1)
# 3. Verify the updated note content
retrieved_note = await nc_client.notes_get_note(note_id=note_id)
retrieved_note = await nc_client.notes.get_note(note_id=note_id)
assert f".attachments.{note_id}/{attachment_filename}" in retrieved_note["content"]
logger.info("Verified image reference exists in updated note content.")
@@ -137,7 +138,7 @@ async def test_note_with_embedded_image(
f"Retrieving image attachment '{attachment_filename}' (category: '{note_category or ''}')..."
)
# Pass category to get_note_attachment
retrieved_img_content, mime_type = await nc_client.get_note_attachment(
retrieved_img_content, mime_type = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=note_category
)
assert retrieved_img_content == image_content
@@ -150,13 +151,13 @@ async def test_note_with_embedded_image(
logger.info(
f"Manually deleting note ID: {note_id} to verify proper attachment cleanup"
)
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Note ID: {note_id} deleted successfully.")
time.sleep(1)
# 6. Verify note is deleted
with pytest.raises(HTTPStatusError) as excinfo_note:
await nc_client.notes_get_note(note_id=note_id)
await nc_client.notes.get_note(note_id=note_id)
assert excinfo_note.value.response.status_code == 404
logger.info(f"Verified note {note_id} deletion (404 received).")
+676
View File
@@ -0,0 +1,676 @@
import json
import logging
import uuid
import pytest
from mcp import ClientSession
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
async def test_mcp_connectivity(nc_mcp_client: ClientSession):
"""Test basic MCP server connectivity and list available tools/resources."""
# List available tools
tools = await nc_mcp_client.list_tools()
logger.info("Available MCP tools:")
tool_names = []
for tool in tools.tools:
logger.info(f" - {tool.name}: {tool.description}")
tool_names.append(tool.name)
# Verify expected tools are present
expected_tools = [
"nc_get_note",
"nc_notes_create_note",
"nc_notes_update_note",
"nc_notes_append_content",
"nc_notes_search_notes",
"nc_notes_delete_note",
"nc_tables_list_tables",
"nc_tables_get_schema",
"nc_tables_read_table",
"nc_tables_insert_row",
"nc_tables_update_row",
"nc_tables_delete_row",
"nc_webdav_list_directory",
"nc_webdav_read_file",
"nc_webdav_write_file",
"nc_webdav_create_directory",
"nc_webdav_delete_resource",
"nc_calendar_list_calendars",
"nc_calendar_create_event",
"nc_calendar_list_events",
"nc_calendar_get_event",
"nc_calendar_update_event",
"nc_calendar_delete_event",
"nc_calendar_create_meeting",
"nc_calendar_get_upcoming_events",
"nc_calendar_find_availability",
"nc_calendar_bulk_operations",
"nc_calendar_manage_calendar",
]
for expected_tool in expected_tools:
assert expected_tool in tool_names, (
f"Expected tool '{expected_tool}' not found in available tools"
)
# List available resource templates
templates = await nc_mcp_client.list_resource_templates()
logger.info("\nAvailable resource templates:")
template_uris = []
for template in templates.resourceTemplates:
logger.info(f" - {template.uriTemplate}")
template_uris.append(template.uriTemplate)
# Verify expected resource templates
expected_templates = ["nc://Notes/{note_id}/attachments/{attachment_filename}"]
for expected_template in expected_templates:
assert expected_template in template_uris, (
f"Expected template '{expected_template}' not found"
)
# List available resources
resources = await nc_mcp_client.list_resources()
logger.info("\nAvailable resources:")
resource_uris = []
for resource in resources.resources:
logger.info(f" - {resource.uri}: {resource.name}")
resource_uris.append(str(resource.uri)) # Convert to string for comparison
# Verify expected resources
expected_resources = ["nc://capabilities", "notes://settings"]
for expected_resource in expected_resources:
assert expected_resource in resource_uris, (
f"Expected resource '{expected_resource}' not found"
)
# List available prompts
prompts = await nc_mcp_client.list_prompts()
logger.info("\nAvailable prompts:")
for prompt in prompts.prompts:
logger.info(f" - {prompt.name}")
async def test_mcp_notes_crud_workflow(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test complete Notes CRUD workflow via MCP tools with verification via NextcloudClient."""
unique_suffix = uuid.uuid4().hex[:8]
test_title = f"MCP Test Note {unique_suffix}"
test_content = f"This is test content for note {unique_suffix}"
test_category = "MCPTesting"
created_note = None
try:
# 1. Create note via MCP
logger.info(f"Creating note via MCP: {test_title}")
create_result = await nc_mcp_client.call_tool(
"nc_notes_create_note",
{"title": test_title, "content": test_content, "category": test_category},
)
assert create_result.isError is False, (
f"MCP note creation failed: {create_result.content}"
)
created_note = create_result.content[0].text
note_data = json.loads(created_note) # Parse the returned JSON
note_id = note_data["id"]
logger.info(f"Note created via MCP with ID: {note_id}")
# 2. Verify creation via direct NextcloudClient
direct_note = await nc_client.notes.get_note(note_id)
assert direct_note["title"] == test_title, (
f"Title mismatch: {direct_note['title']} != {test_title}"
)
assert direct_note["content"] == test_content, "Content mismatch"
assert direct_note["category"] == test_category, "Category mismatch"
# 3. Read note via MCP
logger.info(f"Reading note via MCP: {note_id}")
read_result = await nc_mcp_client.call_tool("nc_get_note", {"note_id": note_id})
assert read_result.isError is False, (
f"MCP note read failed: {read_result.content}"
)
read_note_data = json.loads(read_result.content[0].text)
assert read_note_data["title"] == test_title
assert read_note_data["content"] == test_content
assert read_note_data["category"] == test_category
# 4. Update note via MCP
updated_title = f"Updated {test_title}"
updated_content = f"Updated content: {test_content}"
etag = read_note_data["etag"]
logger.info(f"Updating note via MCP: {note_id}")
update_result = await nc_mcp_client.call_tool(
"nc_notes_update_note",
{
"note_id": note_id,
"etag": etag,
"title": updated_title,
"content": updated_content,
"category": test_category,
},
)
assert update_result.isError is False, (
f"MCP note update failed: {update_result.content}"
)
# 5. Verify update via direct NextcloudClient
updated_direct_note = await nc_client.notes.get_note(note_id)
assert updated_direct_note["title"] == updated_title
assert updated_direct_note["content"] == updated_content
# 6. Append content via MCP
append_content = "\n\nThis is appended content via MCP."
logger.info(f"Appending content to note via MCP: {note_id}")
append_result = await nc_mcp_client.call_tool(
"nc_notes_append_content", {"note_id": note_id, "content": append_content}
)
assert append_result.isError is False, (
f"MCP note append failed: {append_result.content}"
)
# 7. Verify append via direct NextcloudClient
appended_direct_note = await nc_client.notes.get_note(note_id)
assert append_content in appended_direct_note["content"]
# 8. Search for note via MCP
logger.info(f"Searching for note via MCP with query: {unique_suffix}")
search_result = await nc_mcp_client.call_tool(
"nc_notes_search_notes", {"query": unique_suffix}
)
assert search_result.isError is False, (
f"MCP note search failed: {search_result.content}"
)
search_notes_text = search_result.content[0].text
logger.info(f"Search result text: {search_notes_text}")
search_notes = json.loads(search_notes_text)
# Ensure search_notes is a list
if not isinstance(search_notes, list):
logger.warning(
f"Expected search results to be a list, got: {type(search_notes)}"
)
search_notes = [search_notes] if search_notes else []
# Find our note in search results
found_note = None
for note in search_notes:
if isinstance(note, dict) and note.get("id") == note_id:
found_note = note
break
assert found_note is not None, (
f"Created note not found in search results. Search returned: {search_notes}"
)
assert found_note["title"] == updated_title
# 9. Delete note via MCP
logger.info(f"Deleting note via MCP: {note_id}")
delete_result = await nc_mcp_client.call_tool(
"nc_notes_delete_note", {"note_id": note_id}
)
assert delete_result.isError is False, (
f"MCP note deletion failed: {delete_result.content}"
)
# 10. Verify deletion via direct NextcloudClient
try:
await nc_client.notes.get_note(note_id)
pytest.fail("Note should have been deleted but was still found")
except Exception:
# Expected - note should be deleted
logger.info(f"Successfully verified note {note_id} was deleted")
created_note = None # Mark as cleaned up
finally:
# Cleanup in case of test failure
if created_note is not None:
try:
note_data = json.loads(created_note)
await nc_client.notes.delete_note(note_data["id"])
logger.info(f"Cleaned up note {note_data['id']} after test failure")
except Exception as e:
logger.warning(f"Failed to cleanup note: {e}")
async def test_mcp_webdav_workflow(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test WebDAV file operations via MCP tools with verification via NextcloudClient."""
unique_suffix = uuid.uuid4().hex[:8]
test_dir = f"mcp_test_dir_{unique_suffix}"
test_file = f"mcp_test_file_{unique_suffix}.txt"
test_file_path = f"{test_dir}/{test_file}"
test_content = f"This is test content for MCP WebDAV testing {unique_suffix}"
try:
# 1. Create directory via MCP
logger.info(f"Creating directory via MCP: {test_dir}")
create_dir_result = await nc_mcp_client.call_tool(
"nc_webdav_create_directory", {"path": test_dir}
)
assert create_dir_result.isError is False, (
f"MCP directory creation failed: {create_dir_result.content}"
)
# 2. Verify directory creation via direct WebDAV
dir_listing = await nc_client.webdav.list_directory("")
dir_names = [item["name"] for item in dir_listing if item["is_directory"]]
assert test_dir in dir_names, f"Directory {test_dir} not found in root listing"
# 3. Write file via MCP
logger.info(f"Writing file via MCP: {test_file_path}")
write_result = await nc_mcp_client.call_tool(
"nc_webdav_write_file",
{
"path": test_file_path,
"content": test_content,
"content_type": "text/plain",
},
)
assert write_result.isError is False, (
f"MCP file write failed: {write_result.content}"
)
# 4. Verify file creation via direct WebDAV
file_listing = await nc_client.webdav.list_directory(test_dir)
file_names = [item["name"] for item in file_listing if not item["is_directory"]]
assert test_file in file_names, (
f"File {test_file} not found in directory listing"
)
# 5. Read file via MCP
logger.info(f"Reading file via MCP: {test_file_path}")
read_result = await nc_mcp_client.call_tool(
"nc_webdav_read_file", {"path": test_file_path}
)
assert read_result.isError is False, (
f"MCP file read failed: {read_result.content}"
)
read_data = json.loads(read_result.content[0].text)
assert read_data["content"] == test_content, "File content mismatch"
assert read_data["path"] == test_file_path
assert "text/plain" in read_data["content_type"]
# 6. Verify file content via direct WebDAV
direct_content, direct_content_type = await nc_client.webdav.read_file(
test_file_path
)
assert direct_content.decode("utf-8") == test_content
# 7. List directory via MCP
logger.info(f"Listing directory via MCP: {test_dir}")
list_result = await nc_mcp_client.call_tool(
"nc_webdav_list_directory", {"path": test_dir}
)
assert list_result.isError is False, (
f"MCP directory listing failed: {list_result.content}"
)
listing_text = list_result.content[0].text
logger.info(f"Directory listing response: {listing_text}")
listing_data = json.loads(listing_text)
# Ensure listing_data is a list
if not isinstance(listing_data, list):
logger.warning(
f"Expected directory listing to be a list, got: {type(listing_data)}"
)
listing_data = [listing_data] if listing_data else []
# Find our file in the listing
found_file = None
for item in listing_data:
if isinstance(item, dict) and item.get("name") == test_file:
found_file = item
break
assert found_file is not None, (
f"File {test_file} not found in MCP directory listing"
)
assert found_file["is_directory"] is False
assert found_file["size"] == len(test_content.encode("utf-8"))
finally:
# Cleanup
try:
logger.info(f"Cleaning up test file: {test_file_path}")
await nc_mcp_client.call_tool(
"nc_webdav_delete_resource", {"path": test_file_path}
)
logger.info(f"Cleaning up test directory: {test_dir}")
await nc_mcp_client.call_tool(
"nc_webdav_delete_resource", {"path": test_dir}
)
except Exception as e:
logger.warning(f"Failed to cleanup WebDAV resources: {e}")
async def test_mcp_resources_access(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test accessing MCP resources and compare with direct API calls."""
# 1. Test capabilities resource
logger.info("Testing capabilities resource via MCP")
caps_result = await nc_mcp_client.read_resource("nc://capabilities")
assert len(caps_result.contents) == 1
mcp_capabilities = json.loads(caps_result.contents[0].text)
# Compare with direct API call
direct_capabilities = await nc_client.capabilities()
# Basic validation - both should have similar structure
# Both return full OCS response structure
assert "ocs" in mcp_capabilities
assert "data" in mcp_capabilities["ocs"]
assert "version" in mcp_capabilities["ocs"]["data"]
assert "ocs" in direct_capabilities
assert "data" in direct_capabilities["ocs"]
assert "version" in direct_capabilities["ocs"]["data"]
# 2. Test notes settings resource
logger.info("Testing notes settings resource via MCP")
settings_result = await nc_mcp_client.read_resource("notes://settings")
assert len(settings_result.contents) == 1
mcp_settings = json.loads(settings_result.contents[0].text)
# Compare with direct API call
direct_settings = await nc_client.notes.get_settings()
# Both should have settings data
assert isinstance(mcp_settings, dict)
assert isinstance(direct_settings, dict)
logger.info("Successfully verified MCP resources match direct API calls")
async def test_mcp_calendar_workflow(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test complete Calendar workflow via MCP tools with verification via NextcloudClient."""
unique_suffix = uuid.uuid4().hex[:8]
test_event_title = f"MCP Test Event {unique_suffix}"
test_location = f"MCP Test Location {unique_suffix}"
created_event = None
calendar_name = None
try:
# 1. List calendars via MCP
logger.info("Listing calendars via MCP")
calendars_result = await nc_mcp_client.call_tool(
"nc_calendar_list_calendars", {}
)
assert calendars_result.isError is False, (
f"MCP calendar listing failed: {calendars_result.content}"
)
calendars_data = json.loads(calendars_result.content[0].text)
# Debug output to understand the structure
logger.info(f"calendars_data type: {type(calendars_data)}")
logger.info(f"calendars_data content: {calendars_data}")
# Handle the case where MCP tool returns a single dict instead of a list
if isinstance(calendars_data, dict):
# Single calendar returned as dict instead of list
calendar_name = calendars_data["name"]
elif isinstance(calendars_data, list) and calendars_data:
# Normal case - list of calendars
calendar_name = calendars_data[0]["name"]
else:
pytest.skip("No calendars available for testing")
logger.info(f"Using calendar: {calendar_name}")
# 2. Create event via MCP
from datetime import datetime, timedelta
tomorrow = datetime.now() + timedelta(days=1)
start_datetime = tomorrow.strftime("%Y-%m-%dT14:00:00")
end_datetime = tomorrow.strftime("%Y-%m-%dT15:00:00")
event_data = {
"calendar_name": calendar_name,
"title": test_event_title,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"description": f"Test event created via MCP {unique_suffix}",
"location": test_location,
"categories": "testing,mcp",
"status": "CONFIRMED",
"priority": 5,
}
logger.info(f"Creating event via MCP: {test_event_title}")
create_result = await nc_mcp_client.call_tool(
"nc_calendar_create_event", event_data
)
assert create_result.isError is False, (
f"MCP event creation failed: {create_result.content}"
)
created_event_data = json.loads(create_result.content[0].text)
event_uid = created_event_data["uid"]
created_event = {"uid": event_uid, "calendar_name": calendar_name}
logger.info(f"Event created via MCP with UID: {event_uid}")
# 3. Verify creation via direct NextcloudClient
direct_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid)
assert direct_event["title"] == test_event_title
assert direct_event["location"] == test_location
assert "testing" in direct_event.get("categories", "")
# 4. Get event via MCP
logger.info(f"Getting event via MCP: {event_uid}")
get_result = await nc_mcp_client.call_tool(
"nc_calendar_get_event",
{"calendar_name": calendar_name, "event_uid": event_uid},
)
assert get_result.isError is False, (
f"MCP event get failed: {get_result.content}"
)
get_event_data = json.loads(get_result.content[0].text)
assert get_event_data["title"] == test_event_title
assert get_event_data["location"] == test_location
# 5. **TEST nc_calendar_list_events - This is the main tool we're testing**
logger.info("Testing nc_calendar_list_events via MCP")
# Get today and next week for date range
today = datetime.now()
next_week = today + timedelta(days=7)
start_date = today.strftime("%Y-%m-%d")
end_date = next_week.strftime("%Y-%m-%d")
list_events_data = {
"calendar_name": calendar_name,
"start_date": start_date,
"end_date": end_date,
"limit": 50,
"location_contains": "MCP Test",
"title_contains": unique_suffix,
}
list_result = await nc_mcp_client.call_tool(
"nc_calendar_list_events", list_events_data
)
assert list_result.isError is False, (
f"MCP list events failed: {list_result.content}"
)
events_data = json.loads(list_result.content[0].text)
# Debug output to understand what nc_calendar_list_events returns
logger.info(f"list_events result type: {type(events_data)}")
logger.info(f"list_events result content: {events_data}")
# Handle single event returned as dict instead of list (same fix as calendars)
if isinstance(events_data, dict):
# Single event returned as dict instead of list
events_data = [events_data]
assert isinstance(events_data, list), "Expected events list"
# Our created event should be in the list
found_event = None
for event in events_data:
if event.get("uid") == event_uid:
found_event = event
break
assert found_event is not None, (
f"Created event {event_uid} not found in events list"
)
assert found_event["title"] == test_event_title
# 6. Test list events across all calendars
logger.info("Testing nc_calendar_list_events across all calendars")
all_calendars_data = {
"calendar_name": "", # Will be ignored
"search_all_calendars": True,
"start_date": start_date,
"end_date": end_date,
"title_contains": unique_suffix,
}
all_list_result = await nc_mcp_client.call_tool(
"nc_calendar_list_events", all_calendars_data
)
assert all_list_result.isError is False, (
f"MCP list all events failed: {all_list_result.content}"
)
all_events_data = json.loads(all_list_result.content[0].text)
# Handle single event returned as dict instead of list (same fix as calendars)
if isinstance(all_events_data, dict):
# Single event returned as dict instead of list
all_events_data = [all_events_data]
assert isinstance(all_events_data, list), "Expected events list"
# Our event should still be found when searching all calendars
found_in_all = any(event.get("uid") == event_uid for event in all_events_data)
assert found_in_all, "Event not found when searching all calendars"
# 7. Update event via MCP
updated_title = f"Updated {test_event_title}"
updated_description = f"Updated description {unique_suffix}"
update_data = {
"calendar_name": calendar_name,
"event_uid": event_uid,
"title": updated_title,
"description": updated_description,
"priority": 1,
}
logger.info(f"Updating event via MCP: {event_uid}")
update_result = await nc_mcp_client.call_tool(
"nc_calendar_update_event", update_data
)
assert update_result.isError is False, (
f"MCP event update failed: {update_result.content}"
)
# 8. Verify update via direct NextcloudClient
updated_direct_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert updated_direct_event["title"] == updated_title
assert updated_direct_event["description"] == updated_description
assert updated_direct_event["priority"] == 1
# 9. Test upcoming events via MCP
logger.info("Testing nc_calendar_get_upcoming_events via MCP")
upcoming_result = await nc_mcp_client.call_tool(
"nc_calendar_get_upcoming_events",
{"calendar_name": calendar_name, "days_ahead": 7, "limit": 10},
)
assert upcoming_result.isError is False, (
f"MCP upcoming events failed: {upcoming_result.content}"
)
upcoming_events = json.loads(upcoming_result.content[0].text)
# Handle single event returned as dict instead of list (same fix as other tools)
if isinstance(upcoming_events, dict):
# Single event returned as dict instead of list
upcoming_events = [upcoming_events]
assert isinstance(upcoming_events, list), "Expected upcoming events list"
# 10. Delete event via MCP
logger.info(f"Deleting event via MCP: {event_uid}")
delete_result = await nc_mcp_client.call_tool(
"nc_calendar_delete_event",
{"calendar_name": calendar_name, "event_uid": event_uid},
)
assert delete_result.isError is False, (
f"MCP event deletion failed: {delete_result.content}"
)
# 11. Verify deletion via direct NextcloudClient
try:
await nc_client.calendar.get_event(calendar_name, event_uid)
pytest.fail("Event should have been deleted but was still found")
except Exception:
# Expected - event should be deleted
logger.info(f"Successfully verified event {event_uid} was deleted")
created_event = None # Mark as cleaned up
except Exception as e:
if "Calendar app may not be enabled" in str(
e
) or "No calendars available" in str(e):
pytest.skip("Calendar functionality not available for testing")
raise
finally:
# Cleanup in case of test failure
if created_event is not None:
try:
await nc_client.calendar.delete_event(
created_event["calendar_name"], created_event["uid"]
)
logger.info(
f"Cleaned up event {created_event['uid']} after test failure"
)
except Exception as e:
logger.warning(f"Failed to cleanup event: {e}")
+19 -21
View File
@@ -1,7 +1,8 @@
import pytest
import logging
import asyncio
import logging
import uuid # Keep uuid if needed for generating unique data within tests
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
@@ -24,7 +25,7 @@ async def test_notes_api_create_and_read(
note_id = created_note_data["id"]
logger.info(f"Reading note created by fixture, ID: {note_id}")
read_note = await nc_client.notes_get_note(note_id=note_id)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["id"] == note_id
assert read_note["title"] == created_note_data["title"]
@@ -46,7 +47,7 @@ async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict
update_content = f"Updated Content {uuid.uuid4().hex[:8]}"
logger.info(f"Attempting to update note ID: {note_id} with etag: {original_etag}")
updated_note = await nc_client.notes_update_note(
updated_note = await nc_client.notes.update(
note_id=note_id,
etag=original_etag,
title=update_title,
@@ -66,7 +67,7 @@ async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict
# Optional: Verify update by reading again
await asyncio.sleep(1) # Allow potential propagation delay
read_updated_note = await nc_client.notes_get_note(note_id=note_id)
read_updated_note = await nc_client.notes.get_note(note_id=note_id)
assert read_updated_note["title"] == update_title
assert read_updated_note["content"] == update_content
logger.info(f"Successfully updated and verified note ID: {note_id}")
@@ -85,7 +86,7 @@ async def test_notes_api_update_conflict(
# Perform a first update to change the etag
first_update_title = f"First Update {uuid.uuid4().hex[:8]}"
logger.info(f"Performing first update on note ID: {note_id} to change etag.")
first_updated_note = await nc_client.notes_update_note(
first_updated_note = await nc_client.notes.update(
note_id=note_id,
etag=original_etag,
title=first_update_title,
@@ -102,7 +103,7 @@ async def test_notes_api_update_conflict(
f"Attempting second update on note ID: {note_id} with OLD etag: {original_etag}"
)
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.notes_update_note(
await nc_client.notes.update(
note_id=note_id,
etag=original_etag, # Use the stale etag
title="This update should fail due to conflict",
@@ -119,7 +120,7 @@ async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
non_existent_id = 999999999 # Use an ID highly unlikely to exist
logger.info(f"\nAttempting to delete non-existent note ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.notes_delete_note(note_id=non_existent_id)
await nc_client.notes.delete_note(note_id=non_existent_id)
assert excinfo.value.response.status_code == 404
logger.info(
f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404."
@@ -139,7 +140,7 @@ async def test_notes_api_append_content_to_existing_note(
append_text = f"Appended content {uuid.uuid4().hex[:8]}"
logger.info(f"Appending content to note ID: {note_id}")
updated_note = await nc_client.notes_append_content(
updated_note = await nc_client.notes.append_content(
note_id=note_id, content=append_text
)
logger.info(f"Note after append: {updated_note}")
@@ -155,7 +156,7 @@ async def test_notes_api_append_content_to_existing_note(
# Verify by reading the note again
await asyncio.sleep(1) # Allow potential propagation delay
read_note = await nc_client.notes_get_note(note_id=note_id)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == expected_content
logger.info(f"Successfully appended content to note ID: {note_id}")
@@ -169,7 +170,7 @@ async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient
test_category = "Test"
logger.info("Creating empty note for append test")
empty_note = await nc_client.notes_create_note(
empty_note = await nc_client.notes.create_note(
title=test_title,
content="",
category=test_category, # Empty content
@@ -180,7 +181,7 @@ async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient
append_text = f"First content {uuid.uuid4().hex[:8]}"
logger.info(f"Appending content to empty note ID: {note_id}")
updated_note = await nc_client.notes_append_content(
updated_note = await nc_client.notes.append_content(
note_id=note_id, content=append_text
)
@@ -189,14 +190,14 @@ async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient
# Verify by reading the note again
await asyncio.sleep(1)
read_note = await nc_client.notes_get_note(note_id=note_id)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == append_text
logger.info(f"Successfully appended content to empty note ID: {note_id}")
finally:
# Clean up the test note
try:
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Cleaned up test note ID: {note_id}")
except Exception as e:
logger.warning(f"Failed to clean up test note ID: {note_id}: {e}")
@@ -218,7 +219,7 @@ async def test_notes_api_append_content_multiple_times(
logger.info(f"Performing multiple appends to note ID: {note_id}")
# First append
updated_note = await nc_client.notes_append_content(
updated_note = await nc_client.notes.append_content(
note_id=note_id, content=first_append
)
@@ -226,7 +227,7 @@ async def test_notes_api_append_content_multiple_times(
assert updated_note["content"] == expected_content_after_first
# Second append
updated_note = await nc_client.notes_append_content(
updated_note = await nc_client.notes.append_content(
note_id=note_id, content=second_append
)
@@ -237,7 +238,7 @@ async def test_notes_api_append_content_multiple_times(
# Verify by reading the note again
await asyncio.sleep(1)
read_note = await nc_client.notes_get_note(note_id=note_id)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == expected_content_after_second
logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
@@ -250,13 +251,10 @@ async def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudCli
logger.info(f"Attempting to append to non-existent note ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.notes_append_content(
await nc_client.notes.append_content(
note_id=non_existent_id, content="This should fail"
)
assert excinfo.value.response.status_code == 404
logger.info(
f"Appending to non-existent note ID: {non_existent_id} correctly failed with 404."
)
# --- Attachment tests moved to test_attachments.py ---
+535
View File
@@ -0,0 +1,535 @@
import asyncio
import logging
import uuid
from typing import Any, Dict
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration
@pytest.fixture(scope="module")
async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]:
"""
Fixture to get information about the sample table that comes with Nextcloud Tables.
This assumes that the sample table exists in the Nextcloud instance.
"""
logger.info("Looking for sample table in Nextcloud Tables app")
# Get all tables
tables = await nc_client.tables.list_tables()
# Look for a sample table (usually created by default)
sample_table = None
for table in tables:
# Common names for sample tables
if any(
keyword in table.get("title", "").lower()
for keyword in ["sample", "demo", "example", "test"]
):
sample_table = table
break
if not sample_table and tables:
# If no sample table found, use the first available table
sample_table = tables[0]
logger.info(
f"No sample table found, using first available table: {sample_table.get('title')}"
)
if not sample_table:
pytest.skip(
"No tables found in Nextcloud Tables app. Please ensure Tables app is installed and has at least one table."
)
# Get the schema for the sample table
table_id = sample_table["id"]
schema = await nc_client.tables.get_table_schema(table_id)
logger.info(f"Using sample table: {sample_table.get('title')} (ID: {table_id})")
return {
"table": sample_table,
"schema": schema,
"table_id": table_id,
"columns": schema.get("columns", []),
}
@pytest.fixture
async def temporary_table_row(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Fixture to create a temporary row in the sample table for testing.
Yields the created row data and cleans up afterward.
"""
table_id = sample_table_info["table_id"]
columns = sample_table_info["columns"]
# Create test data based on the table schema
test_data = {}
unique_suffix = uuid.uuid4().hex[:8]
for column in columns:
column_id = column["id"]
column_type = column.get("type", "text")
column_title = column.get("title", f"column_{column_id}")
# Generate test data based on column type
if column_type == "text":
test_data[column_id] = f"Test {column_title} {unique_suffix}"
elif column_type == "number":
test_data[column_id] = 42
elif column_type == "datetime":
test_data[column_id] = "2024-01-01T12:00:00Z"
elif column_type == "select":
# For select columns, use the first option if available
options = column.get("selectOptions", [])
if options:
test_data[column_id] = options[0].get("label", "Option 1")
else:
test_data[column_id] = "Test Option"
else:
# Default to text for unknown types
test_data[column_id] = f"Test {column_title} {unique_suffix}"
logger.info(f"Creating temporary row in table {table_id} with data: {test_data}")
created_row = None
try:
created_row = await nc_client.tables.create_row(table_id, test_data)
row_id = created_row.get("id")
if not row_id:
pytest.fail("Failed to get ID from created temporary row.")
logger.info(f"Temporary row created with ID: {row_id}")
yield created_row
finally:
if created_row and created_row.get("id"):
row_id = created_row["id"]
logger.info(f"Cleaning up temporary row ID: {row_id}")
try:
await nc_client.tables.delete_row(row_id)
logger.info(f"Successfully deleted temporary row ID: {row_id}")
except HTTPStatusError as e:
# Ignore 404 if row was already deleted by the test itself
if e.response.status_code != 404:
logger.error(f"HTTP error deleting temporary row {row_id}: {e}")
else:
logger.warning(f"Temporary row {row_id} already deleted (404).")
except Exception as e:
logger.error(f"Unexpected error deleting temporary row {row_id}: {e}")
async def test_tables_list_tables(nc_client: NextcloudClient):
"""
Test listing all tables available to the user.
"""
logger.info("Testing list_tables functionality")
tables = await nc_client.tables.list_tables()
assert isinstance(tables, list)
assert len(tables) > 0, "Expected at least one table to be available"
# Check that each table has required fields
for table in tables:
assert "id" in table
assert "title" in table
assert isinstance(table["id"], int)
assert isinstance(table["title"], str)
logger.info(f"Successfully listed {len(tables)} tables")
async def test_tables_get_schema(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Test getting the schema/structure of a specific table.
"""
table_id = sample_table_info["table_id"]
logger.info(f"Testing get_table_schema for table ID: {table_id}")
schema = await nc_client.tables.get_table_schema(table_id)
assert isinstance(schema, dict)
assert "columns" in schema
assert isinstance(schema["columns"], list)
assert len(schema["columns"]) > 0, "Expected at least one column in the table"
# Check that each column has required fields
for column in schema["columns"]:
assert "id" in column
assert "title" in column
assert "type" in column
assert isinstance(column["id"], int)
assert isinstance(column["title"], str)
assert isinstance(column["type"], str)
logger.info(f"Successfully retrieved schema with {len(schema['columns'])} columns")
async def test_tables_read_table(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Test reading rows from a table.
"""
table_id = sample_table_info["table_id"]
logger.info(f"Testing get_table_rows for table ID: {table_id}")
# Test without pagination
rows = await nc_client.tables.get_table_rows(table_id)
assert isinstance(rows, list)
# Note: The table might be empty, so we don't assert len > 0
# Test with pagination
rows_limited = await nc_client.tables.get_table_rows(table_id, limit=5, offset=0)
assert isinstance(rows_limited, list)
assert len(rows_limited) <= 5
# If there are rows, check their structure
if rows:
row = rows[0]
assert "id" in row
assert "tableId" in row
assert "data" in row
assert isinstance(row["id"], int)
assert isinstance(row["tableId"], int)
assert isinstance(row["data"], list)
logger.info(f"Successfully read {len(rows)} rows from table")
async def test_tables_create_row(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Test creating a new row in a table.
"""
table_id = sample_table_info["table_id"]
columns = sample_table_info["columns"]
# Create test data based on the table schema
test_data = {}
unique_suffix = uuid.uuid4().hex[:8]
for column in columns:
column_id = column["id"]
column_type = column.get("type", "text")
column_title = column.get("title", f"column_{column_id}")
# Generate test data based on column type
if column_type == "text":
test_data[column_id] = f"Test Create {column_title} {unique_suffix}"
elif column_type == "number":
test_data[column_id] = 123
elif column_type == "datetime":
test_data[column_id] = "2024-01-01T12:00:00Z"
elif column_type == "select":
# For select columns, use the first option if available
options = column.get("selectOptions", [])
if options:
test_data[column_id] = options[0].get("label", "Option 1")
else:
test_data[column_id] = "Test Option"
else:
# Default to text for unknown types
test_data[column_id] = f"Test Create {column_title} {unique_suffix}"
logger.info(f"Testing create_row for table ID: {table_id} with data: {test_data}")
created_row = None
try:
created_row = await nc_client.tables.create_row(table_id, test_data)
assert isinstance(created_row, dict)
assert "id" in created_row
assert "tableId" in created_row
assert isinstance(created_row["id"], int)
assert created_row["tableId"] == table_id
# Verify the row was created by reading it back
await asyncio.sleep(1) # Allow potential propagation delay
rows = await nc_client.tables.get_table_rows(table_id)
created_row_id = created_row["id"]
# Find the created row in the results
found_row = None
for row in rows:
if row["id"] == created_row_id:
found_row = row
break
assert found_row is not None, (
f"Created row with ID {created_row_id} not found in table"
)
logger.info(f"Successfully created row with ID: {created_row_id}")
finally:
# Clean up the created row
if created_row and created_row.get("id"):
try:
await nc_client.tables.delete_row(created_row["id"])
logger.info(f"Cleaned up created row ID: {created_row['id']}")
except Exception as e:
logger.warning(f"Failed to clean up created row: {e}")
async def test_tables_update_row(
nc_client: NextcloudClient,
temporary_table_row: Dict[str, Any],
sample_table_info: Dict[str, Any],
):
"""
Test updating an existing row in a table.
"""
row_id = temporary_table_row["id"]
columns = sample_table_info["columns"]
# Create updated data
update_data = {}
unique_suffix = uuid.uuid4().hex[:8]
for column in columns:
column_id = column["id"]
column_type = column.get("type", "text")
column_title = column.get("title", f"column_{column_id}")
# Generate updated test data based on column type
if column_type == "text":
update_data[column_id] = f"Updated {column_title} {unique_suffix}"
elif column_type == "number":
update_data[column_id] = 456
elif column_type == "datetime":
update_data[column_id] = "2024-12-31T23:59:59Z"
elif column_type == "select":
# For select columns, use the first option if available
options = column.get("selectOptions", [])
if options:
update_data[column_id] = options[0].get("label", "Option 1")
else:
update_data[column_id] = "Updated Option"
else:
# Default to text for unknown types
update_data[column_id] = f"Updated {column_title} {unique_suffix}"
logger.info(f"Testing update_row for row ID: {row_id} with data: {update_data}")
updated_row = await nc_client.tables.update_row(row_id, update_data)
assert isinstance(updated_row, dict)
assert "id" in updated_row
assert updated_row["id"] == row_id
# Verify the row was updated by reading it back
await asyncio.sleep(1) # Allow potential propagation delay
table_id = sample_table_info["table_id"]
rows = await nc_client.tables.get_table_rows(table_id)
# Find the updated row in the results
found_row = None
for row in rows:
if row["id"] == row_id:
found_row = row
break
assert found_row is not None, f"Updated row with ID {row_id} not found in table"
logger.info(f"Successfully updated row with ID: {row_id}")
async def test_tables_delete_row(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Test deleting a row from a table.
"""
table_id = sample_table_info["table_id"]
columns = sample_table_info["columns"]
# First create a row to delete
test_data = {}
unique_suffix = uuid.uuid4().hex[:8]
for column in columns:
column_id = column["id"]
column_type = column.get("type", "text")
column_title = column.get("title", f"column_{column_id}")
if column_type == "text":
test_data[column_id] = f"Test Delete {column_title} {unique_suffix}"
elif column_type == "number":
test_data[column_id] = 789
elif column_type == "datetime":
test_data[column_id] = "2024-06-15T10:30:00Z"
elif column_type == "select":
options = column.get("selectOptions", [])
if options:
test_data[column_id] = options[0].get("label", "Option 1")
else:
test_data[column_id] = "Delete Option"
else:
test_data[column_id] = f"Test Delete {column_title} {unique_suffix}"
logger.info(f"Creating row for delete test in table ID: {table_id}")
created_row = await nc_client.tables.create_row(table_id, test_data)
row_id = created_row["id"]
logger.info(f"Testing delete_row for row ID: {row_id}")
# Delete the row
delete_result = await nc_client.tables.delete_row(row_id)
assert isinstance(delete_result, dict)
# The delete response might vary, but it should be successful
# Verify the row was deleted by trying to find it
await asyncio.sleep(1) # Allow potential propagation delay
rows = await nc_client.tables.get_table_rows(table_id)
# Ensure the deleted row is not in the results
found_row = None
for row in rows:
if row["id"] == row_id:
found_row = row
break
assert found_row is None, f"Deleted row with ID {row_id} still found in table"
logger.info(f"Successfully deleted row with ID: {row_id}")
async def test_tables_delete_nonexistent_row(nc_client: NextcloudClient):
"""
Test that deleting a non-existent row fails appropriately.
"""
non_existent_id = 999999999 # Use an ID highly unlikely to exist
logger.info(f"Testing delete_row for non-existent row ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.tables.delete_row(non_existent_id)
# Accept both 404 and 500 as valid error responses for non-existent rows
# The API behavior may vary between Nextcloud versions
assert excinfo.value.response.status_code in [404, 500]
logger.info(
f"Deleting non-existent row ID: {non_existent_id} correctly failed with {excinfo.value.response.status_code}."
)
async def test_tables_transform_row_data(
nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
):
"""
Test the transform_row_data utility method.
"""
table_id = sample_table_info["table_id"]
columns = sample_table_info["columns"]
logger.info(f"Testing transform_row_data for table ID: {table_id}")
# Get some rows to transform
rows = await nc_client.tables.get_table_rows(table_id, limit=5)
if not rows:
logger.info("No rows to transform, skipping transform_row_data test")
return
# Transform the rows
transformed_rows = nc_client.tables.transform_row_data(rows, columns)
assert isinstance(transformed_rows, list)
assert len(transformed_rows) == len(rows)
# Check the structure of transformed rows
for i, transformed_row in enumerate(transformed_rows):
original_row = rows[i]
assert "id" in transformed_row
assert "tableId" in transformed_row
assert "data" in transformed_row
assert transformed_row["id"] == original_row["id"]
assert transformed_row["tableId"] == original_row["tableId"]
assert isinstance(transformed_row["data"], dict)
# Check that column IDs were transformed to column names
for column in columns:
column_title = column["title"]
# The transformed data should have column names as keys
# (though the column might not have data in this row)
if any(item["columnId"] == column["id"] for item in original_row["data"]):
assert column_title in transformed_row["data"]
logger.info(f"Successfully transformed {len(transformed_rows)} rows")
async def test_tables_get_nonexistent_table_schema(nc_client: NextcloudClient):
"""
Test that getting schema for a non-existent table fails appropriately.
"""
non_existent_id = 999999999 # Use an ID highly unlikely to exist
logger.info(
f"Testing get_table_schema for non-existent table ID: {non_existent_id}"
)
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.tables.get_table_schema(non_existent_id)
assert excinfo.value.response.status_code == 404
logger.info(
f"Getting schema for non-existent table ID: {non_existent_id} correctly failed with 404."
)
async def test_tables_read_nonexistent_table(nc_client: NextcloudClient):
"""
Test that reading from a non-existent table fails appropriately.
"""
non_existent_id = 999999999 # Use an ID highly unlikely to exist
logger.info(f"Testing get_table_rows for non-existent table ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.tables.get_table_rows(non_existent_id)
assert excinfo.value.response.status_code == 404
logger.info(
f"Reading from non-existent table ID: {non_existent_id} correctly failed with 404."
)
async def test_tables_create_row_invalid_table(nc_client: NextcloudClient):
"""
Test that creating a row in a non-existent table fails appropriately.
"""
non_existent_id = 999999999 # Use an ID highly unlikely to exist
test_data = {1: "test value"}
logger.info(f"Testing create_row for non-existent table ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.tables.create_row(non_existent_id, test_data)
assert excinfo.value.response.status_code == 404
logger.info(
f"Creating row in non-existent table ID: {non_existent_id} correctly failed with 404."
)
+12 -11
View File
@@ -1,7 +1,8 @@
import pytest
import logging
import time
import uuid
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
@@ -29,7 +30,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
try:
# 1. Create note with initial category
logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
created_note = await nc_client.notes_create_note(
created_note = await nc_client.notes.create_note(
title=note_title, content="Initial content", category=initial_category
)
note_id = created_note["id"]
@@ -41,7 +42,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
logger.info(
f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
)
upload_response = await nc_client.add_note_attachment(
upload_response = await nc_client.webdav.add_note_attachment(
note_id=note_id,
filename=attachment_filename,
content=attachment_content,
@@ -56,7 +57,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
logger.info(
f"Verifying attachment retrieval from initial category '{initial_category}'"
)
retrieved_content1, _ = await nc_client.get_note_attachment(
retrieved_content1, _ = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category
)
assert retrieved_content1 == attachment_content
@@ -72,9 +73,9 @@ async def test_category_change_cleans_up_old_attachments_directory(
logger.info(
f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
)
current_note_data = await nc_client.notes_get_note(note_id=note_id)
current_note_data = await nc_client.notes.get_note(note_id=note_id)
current_etag = current_note_data["etag"]
updated_note = await nc_client.notes_update_note(
updated_note = await nc_client.notes.update(
note_id=note_id,
etag=current_etag,
category=new_category,
@@ -90,7 +91,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
logger.info(
f"Verifying attachment retrieval from new category '{new_category}'"
)
retrieved_content2, _ = await nc_client.get_note_attachment(
retrieved_content2, _ = await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=new_category
)
assert retrieved_content2 == attachment_content
@@ -101,7 +102,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
f"Trying to retrieve attachment from old category '{initial_category}' - should fail"
)
try:
await nc_client.get_note_attachment(
await nc_client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category
)
# If we get here, it means the old directory still exists (a problem)
@@ -165,14 +166,14 @@ async def test_category_change_cleans_up_old_attachments_directory(
if note_id:
logger.info(f"Cleaning up note ID: {note_id}")
try:
await nc_client.notes_delete_note(note_id=note_id)
await nc_client.notes.delete_note(note_id=note_id)
logger.info(f"Note {note_id} deleted.")
time.sleep(1)
# 9. Verify both old and new attachment paths are gone
logger.info("Verifying all attachment paths are gone")
with pytest.raises(HTTPStatusError) as excinfo_new:
await nc_client.get_note_attachment(
await nc_client.webdav.get_note_attachment(
note_id=note_id,
filename=attachment_filename,
category=new_category,
@@ -180,7 +181,7 @@ async def test_category_change_cleans_up_old_attachments_directory(
assert excinfo_new.value.response.status_code == 404
with pytest.raises(HTTPStatusError) as excinfo_old:
await nc_client.get_note_attachment(
await nc_client.webdav.get_note_attachment(
note_id=note_id,
filename=attachment_filename,
category=initial_category,
+273
View File
@@ -0,0 +1,273 @@
"""Integration tests for WebDAV operations."""
import logging
import uuid
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration
@pytest.fixture
async def test_base_path(nc_client: NextcloudClient):
"""Base path for test files/directories."""
test_dir = f"mcp_test_{uuid.uuid4().hex[:8]}"
await nc_client.webdav.create_directory(test_dir)
yield test_dir
await nc_client.webdav.delete_resource(test_dir)
async def test_create_and_delete_directory(
nc_client: NextcloudClient, test_base_path: str
):
"""Test creating and deleting directories."""
test_dir = f"{test_base_path}/test_directory"
try:
# Create directory
result = await nc_client.webdav.create_directory(test_dir)
assert result["status_code"] == 201 # Created
logger.info(f"Created directory: {test_dir}")
# Verify directory exists by listing parent
parent_listing = await nc_client.webdav.list_directory(test_base_path)
dir_names = [item["name"] for item in parent_listing]
assert "test_directory" in dir_names
# Delete directory
delete_result = await nc_client.webdav.delete_resource(test_dir)
assert delete_result["status_code"] in [204, 404] # No Content or Not Found
logger.info(f"Deleted directory: {test_dir}")
finally:
# Cleanup: ensure directory is deleted
try:
await nc_client.webdav.delete_resource(test_dir)
except Exception:
pass
async def test_write_read_delete_file(nc_client: NextcloudClient, test_base_path: str):
"""Test writing, reading, and deleting files."""
test_file = f"{test_base_path}/test_file.txt"
test_content = f"Test content {uuid.uuid4().hex}"
try:
# Create base directory first
await nc_client.webdav.create_directory(test_base_path)
# Write file
write_result = await nc_client.webdav.write_file(
test_file, test_content.encode("utf-8"), content_type="text/plain"
)
assert write_result["status_code"] in [200, 201, 204] # Success codes
logger.info(f"Wrote file: {test_file}")
# Read file back
content, content_type = await nc_client.webdav.read_file(test_file)
assert content.decode("utf-8") == test_content
assert "text/plain" in content_type
logger.info(f"Read file: {test_file}")
# Verify file appears in directory listing
listing = await nc_client.webdav.list_directory(test_base_path)
file_names = [item["name"] for item in listing]
assert "test_file.txt" in file_names
# Delete file
delete_result = await nc_client.webdav.delete_resource(test_file)
assert delete_result["status_code"] in [204, 404] # No Content or Not Found
logger.info(f"Deleted file: {test_file}")
finally:
# Cleanup
try:
await nc_client.webdav.delete_resource(test_file)
await nc_client.webdav.delete_resource(test_base_path)
except Exception:
pass
async def test_list_directory_empty_and_populated(
nc_client: NextcloudClient, test_base_path: str
):
"""Test listing empty and populated directories."""
try:
# Create base directory
await nc_client.webdav.create_directory(test_base_path)
# List empty directory
empty_listing = await nc_client.webdav.list_directory(test_base_path)
assert isinstance(empty_listing, list)
assert len(empty_listing) == 0
logger.info(f"Empty directory listing: {len(empty_listing)} items")
# Add some files and directories
await nc_client.webdav.create_directory(f"{test_base_path}/subdir1")
await nc_client.webdav.create_directory(f"{test_base_path}/subdir2")
await nc_client.webdav.write_file(
f"{test_base_path}/file1.txt", b"content1", content_type="text/plain"
)
await nc_client.webdav.write_file(
f"{test_base_path}/file2.md",
b"# Markdown content",
content_type="text/markdown",
)
# List populated directory
populated_listing = await nc_client.webdav.list_directory(test_base_path)
assert len(populated_listing) == 4 # 2 dirs + 2 files
# Check that we have both files and directories
names = [item["name"] for item in populated_listing]
assert "subdir1" in names
assert "subdir2" in names
assert "file1.txt" in names
assert "file2.md" in names
# Check metadata is present
for item in populated_listing:
assert "name" in item
assert "path" in item
assert "is_directory" in item
assert "size" in item
assert "content_type" in item
assert "last_modified" in item
logger.info(f"Populated directory listing: {len(populated_listing)} items")
finally:
# Cleanup
try:
await nc_client.webdav.delete_resource(f"{test_base_path}/file1.txt")
await nc_client.webdav.delete_resource(f"{test_base_path}/file2.md")
await nc_client.webdav.delete_resource(f"{test_base_path}/subdir1")
await nc_client.webdav.delete_resource(f"{test_base_path}/subdir2")
await nc_client.webdav.delete_resource(test_base_path)
except Exception:
pass
async def test_read_nonexistent_file(nc_client: NextcloudClient):
"""Test reading a file that doesn't exist."""
nonexistent_file = f"nonexistent_{uuid.uuid4().hex}.txt"
with pytest.raises(HTTPStatusError) as exc_info:
await nc_client.webdav.read_file(nonexistent_file)
assert exc_info.value.response.status_code == 404
logger.info(f"Correctly got 404 for nonexistent file: {nonexistent_file}")
async def test_delete_nonexistent_resource(nc_client: NextcloudClient):
"""Test deleting a resource that doesn't exist."""
nonexistent_resource = f"nonexistent_{uuid.uuid4().hex}"
result = await nc_client.webdav.delete_resource(nonexistent_resource)
assert result["status_code"] == 404
logger.info(f"Correctly got 404 for nonexistent resource: {nonexistent_resource}")
async def test_create_nested_directories(
nc_client: NextcloudClient, test_base_path: str
):
"""Test creating nested directory structures."""
nested_path = f"{test_base_path}/level1/level2/level3"
try:
# Create nested directories (should create parent directories automatically)
result = await nc_client.webdav.create_directory(nested_path, True)
assert result["status_code"] == 201
# Verify the structure was created
level1_listing = await nc_client.webdav.list_directory(
f"{test_base_path}/level1"
)
assert len(level1_listing) == 1
assert level1_listing[0]["name"] == "level2"
assert level1_listing[0]["is_directory"] is True
level2_listing = await nc_client.webdav.list_directory(
f"{test_base_path}/level1/level2"
)
assert len(level2_listing) == 1
assert level2_listing[0]["name"] == "level3"
assert level2_listing[0]["is_directory"] is True
logger.info(f"Created nested directory structure: {nested_path}")
finally:
# Cleanup - delete from deepest to shallowest
try:
await nc_client.webdav.delete_resource(nested_path)
await nc_client.webdav.delete_resource(f"{test_base_path}/level1/level2")
await nc_client.webdav.delete_resource(f"{test_base_path}/level1")
except Exception:
pass
async def test_overwrite_existing_file(nc_client: NextcloudClient, test_base_path: str):
"""Test overwriting an existing file."""
test_file = f"{test_base_path}/overwrite_test.txt"
original_content = "Original content"
new_content = "New content after overwrite"
try:
# Create base directory
await nc_client.webdav.create_directory(test_base_path)
# Write original file
await nc_client.webdav.write_file(
test_file, original_content.encode("utf-8"), content_type="text/plain"
)
# Verify original content
content, _ = await nc_client.webdav.read_file(test_file)
assert content.decode("utf-8") == original_content
# Overwrite with new content
overwrite_result = await nc_client.webdav.write_file(
test_file, new_content.encode("utf-8"), content_type="text/plain"
)
assert overwrite_result["status_code"] in [200, 204] # OK or No Content
# Verify new content
content, _ = await nc_client.webdav.read_file(test_file)
assert content.decode("utf-8") == new_content
logger.info(f"Successfully overwrote file: {test_file}")
finally:
# Cleanup
try:
await nc_client.webdav.delete_resource(test_file)
await nc_client.webdav.delete_resource(test_base_path)
except Exception:
pass
async def test_list_root_directory(nc_client: NextcloudClient):
"""Test listing the root directory."""
root_listing = await nc_client.webdav.list_directory("")
# Root directory should exist and be listable
assert isinstance(root_listing, list)
# Should have at least some default folders/files
assert len(root_listing) >= 0
# Check structure of items
for item in root_listing:
assert "name" in item
assert "path" in item
assert "is_directory" in item
assert "size" in item
assert "content_type" in item
assert "last_modified" in item
logger.info(f"Root directory contains {len(root_listing)} items")
Generated
+189 -8
View File
@@ -43,6 +43,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/25/8a/c46dcc25341b5bce5472c718902eb3d38600a903b14fa6aeecef3f21a46f/asttokens-3.0.0-py3-none-any.whl", hash = "sha256:e3078351a059199dd5138cb1c706e6430c05eff2ff136af5eb4790f9d28932e2", size = 26918, upload-time = "2024-11-30T04:30:10.946Z" },
]
[[package]]
name = "attrs"
version = "25.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/5a/b0/1367933a8532ee6ff8d63537de4f1177af4bff9f3e829baf7331f595bb24/attrs-25.3.0.tar.gz", hash = "sha256:75d7cefc7fb576747b2c81b4442d4d4a1ce0900973527c011d1030fd3bf4af1b", size = 812032, upload-time = "2025-03-13T11:10:22.779Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/77/06/bb80f5f86020c4551da315d78b3ab75e8228f89f0162f2c3a819e407941a/attrs-25.3.0-py3-none-any.whl", hash = "sha256:427318ce031701fea540783410126f03899a97ffc6f61596ad581ac2e40e3bc3", size = 63815, upload-time = "2025-03-13T11:10:21.14Z" },
]
[[package]]
name = "certifi"
version = "2025.4.26"
@@ -270,6 +279,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e1/9b/a181f281f65d776426002f330c31849b86b31fc9d848db62e16f03ff739f/httpx_sse-0.4.0-py3-none-any.whl", hash = "sha256:f329af6eae57eaa2bdfd962b42524764af68075ea87370a2de920af5341e318f", size = 7819, upload-time = "2023-12-22T08:01:19.89Z" },
]
[[package]]
name = "icalendar"
version = "6.3.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "python-dateutil" },
{ name = "tzdata" },
]
sdist = { url = "https://files.pythonhosted.org/packages/08/13/e5899c916dcf1343ea65823eb7278d3e1a1d679f383f6409380594b5f322/icalendar-6.3.1.tar.gz", hash = "sha256:a697ce7b678072941e519f2745704fc29d78ef92a2dc53d9108ba6a04aeba466", size = 177169, upload-time = "2025-05-20T07:42:50.683Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/6c/25/b5fc00e85d2dfaf5c806ac8b5f1de072fa11630c5b15b4ae5bbc228abd51/icalendar-6.3.1-py3-none-any.whl", hash = "sha256:7ea1d1b212df685353f74cdc6ec9646bf42fa557d1746ea645ce8779fdfbecdd", size = 242349, upload-time = "2025-05-20T07:42:48.589Z" },
]
[[package]]
name = "idna"
version = "3.10"
@@ -346,6 +368,33 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" },
]
[[package]]
name = "jsonschema"
version = "4.24.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "jsonschema-specifications" },
{ name = "referencing" },
{ name = "rpds-py" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bf/d3/1cf5326b923a53515d8f3a2cd442e6d7e94fcc444716e879ea70a0ce3177/jsonschema-4.24.0.tar.gz", hash = "sha256:0b4e8069eb12aedfa881333004bccaec24ecef5a8a6a4b6df142b2cc9599d196", size = 353480, upload-time = "2025-05-26T18:48:10.459Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a2/3d/023389198f69c722d039351050738d6755376c8fd343e91dc493ea485905/jsonschema-4.24.0-py3-none-any.whl", hash = "sha256:a462455f19f5faf404a7902952b6f0e3ce868f3ee09a359b05eca6673bd8412d", size = 88709, upload-time = "2025-05-26T18:48:08.417Z" },
]
[[package]]
name = "jsonschema-specifications"
version = "2025.4.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "referencing" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bf/ce/46fbd9c8119cfc3581ee5643ea49464d168028cfb5caff5fc0596d0cf914/jsonschema_specifications-2025.4.1.tar.gz", hash = "sha256:630159c9f4dbea161a6a2205c3011cc4f18ff381b189fff48bb39b9bf26ae608", size = 15513, upload-time = "2025-04-23T12:34:07.418Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/01/0e/b27cdbaccf30b890c40ed1da9fd4a3593a5cf94dae54fb34f8a4b74fcd3f/jsonschema_specifications-2025.4.1-py3-none-any.whl", hash = "sha256:4653bffbd6584f7de83a67e0d620ef16900b390ddc7939d56684d6c81e33f1af", size = 18437, upload-time = "2025-04-23T12:34:05.422Z" },
]
[[package]]
name = "markdown-it-py"
version = "3.0.0"
@@ -420,12 +469,13 @@ wheels = [
[[package]]
name = "mcp"
version = "1.9.0"
version = "1.10.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "httpx" },
{ name = "httpx-sse" },
{ name = "jsonschema" },
{ name = "pydantic" },
{ name = "pydantic-settings" },
{ name = "python-multipart" },
@@ -433,9 +483,9 @@ dependencies = [
{ name = "starlette" },
{ name = "uvicorn", marker = "sys_platform != 'emscripten'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bc/8d/0f4468582e9e97b0a24604b585c651dfd2144300ecffd1c06a680f5c8861/mcp-1.9.0.tar.gz", hash = "sha256:905d8d208baf7e3e71d70c82803b89112e321581bcd2530f9de0fe4103d28749", size = 281432, upload-time = "2025-05-15T18:51:06.615Z" }
sdist = { url = "https://files.pythonhosted.org/packages/7c/68/63045305f29ff680a9cd5be360c755270109e6b76f696ea6824547ddbc30/mcp-1.10.1.tar.gz", hash = "sha256:aaa0957d8307feeff180da2d9d359f2b801f35c0c67f1882136239055ef034c2", size = 392969, upload-time = "2025-06-27T12:03:08.982Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a5/d5/22e36c95c83c80eb47c83f231095419cf57cf5cca5416f1c960032074c78/mcp-1.9.0-py3-none-any.whl", hash = "sha256:9dfb89c8c56f742da10a5910a1f64b0d2ac2c3ed2bd572ddb1cfab7f35957178", size = 125082, upload-time = "2025-05-15T18:51:04.916Z" },
{ url = "https://files.pythonhosted.org/packages/d7/3f/435a5b3d10ae242a9d6c2b33175551173c3c61fe637dc893be05c4ed0aaf/mcp-1.10.1-py3-none-any.whl", hash = "sha256:4d08301aefe906dce0fa482289db55ce1db831e3e67212e65b5e23ad8454b3c5", size = 150878, upload-time = "2025-06-27T12:03:07.328Z" },
]
[package.optional-dependencies]
@@ -455,12 +505,14 @@ wheels = [
[[package]]
name = "nextcloud-mcp-server"
version = "0.2.5"
version = "0.6.1"
source = { editable = "." }
dependencies = [
{ name = "httpx" },
{ name = "icalendar" },
{ name = "mcp", extra = ["cli"] },
{ name = "pillow" },
{ name = "pythonvcard4" },
]
[package.dev-dependencies]
@@ -476,8 +528,10 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "httpx", specifier = ">=0.28.1,<0.29.0" },
{ name = "mcp", extras = ["cli"], specifier = ">=1.9,<1.10" },
{ name = "icalendar", specifier = ">=6.0.0,<7.0.0" },
{ name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" },
{ name = "pillow", specifier = ">=11.2.1,<12.0.0" },
{ name = "pythonvcard4", specifier = ">=0.2.0" },
]
[package.metadata.requires-dev]
@@ -761,6 +815,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/28/d0/def53b4a790cfb21483016430ed828f64830dd981ebe1089971cd10cab25/pytest_cov-6.1.1-py3-none-any.whl", hash = "sha256:bddf29ed2d0ab6f4df17b4c55b0a657287db8684af9c42ea546b21b1041b3dde", size = 23841, upload-time = "2025-04-05T14:07:49.641Z" },
]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "six" },
]
sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
]
[[package]]
name = "python-dotenv"
version = "1.1.0"
@@ -779,6 +845,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/45/58/38b5afbc1a800eeea951b9285d3912613f2603bdf897a4ab0f4bd7f405fc/python_multipart-0.0.20-py3-none-any.whl", hash = "sha256:8a62d3a8335e06589fe01f2a3e178cdcc632f3fbe0d492ad9ee0ec35aab1f104", size = 24546, upload-time = "2024-12-16T19:45:44.423Z" },
]
[[package]]
name = "pythonvcard4"
version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/34/04/02d5952a9d8cbcb9e62b4fc4f6f842e8d43aead6e307f83e6fd6f7352fbd/pythonvcard4-0.2.0.tar.gz", hash = "sha256:236bba2769e459645cfa776407ff07856aced45b437116bf40ddb39bbcefdb6d", size = 5530, upload-time = "2025-04-26T23:18:48.963Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/2f/ee10d88bbe12e4e9e06f81589d999687038e5cd5fec6c05aed57c50aede6/pythonvcard4-0.2.0-py3-none-any.whl", hash = "sha256:dce31355dd50aee537f8883de86f301510e407bc1755a68ec8d5055b64f5c660", size = 5890, upload-time = "2025-04-26T23:18:48.2Z" },
]
[[package]]
name = "pyyaml"
version = "6.0.2"
@@ -826,6 +901,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ad/3f/11dd4cd4f39e05128bfd20138faea57bec56f9ffba6185d276e3107ba5b2/questionary-2.1.0-py3-none-any.whl", hash = "sha256:44174d237b68bc828e4878c763a9ad6790ee61990e0ae72927694ead57bab8ec", size = 36747, upload-time = "2024-12-29T11:49:16.734Z" },
]
[[package]]
name = "referencing"
version = "0.36.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "rpds-py" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/2f/db/98b5c277be99dd18bfd91dd04e1b759cad18d1a338188c936e92f921c7e2/referencing-0.36.2.tar.gz", hash = "sha256:df2e89862cd09deabbdba16944cc3f10feb6b3e6f18e902f7cc25609a34775aa", size = 74744, upload-time = "2025-01-25T08:48:16.138Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c1/b1/3baf80dc6d2b7bc27a95a67752d0208e410351e3feb4eb78de5f77454d8d/referencing-0.36.2-py3-none-any.whl", hash = "sha256:e8699adbbf8b5c7de96d8ffa0eb5c158b3beafce084968e2ea8bb08c6794dcd0", size = 26775, upload-time = "2025-01-25T08:48:14.241Z" },
]
[[package]]
name = "rich"
version = "14.0.0"
@@ -839,6 +928,80 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/9b/63f4c7ebc259242c89b3acafdb37b41d1185c07ff0011164674e9076b491/rich-14.0.0-py3-none-any.whl", hash = "sha256:1c9491e1951aac09caffd42f448ee3d04e58923ffe14993f6e83068dc395d7e0", size = 243229, upload-time = "2025-03-30T14:15:12.283Z" },
]
[[package]]
name = "rpds-py"
version = "0.25.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8c/a6/60184b7fc00dd3ca80ac635dd5b8577d444c57e8e8742cecabfacb829921/rpds_py-0.25.1.tar.gz", hash = "sha256:8960b6dac09b62dac26e75d7e2c4a22efb835d827a7278c34f72b2b84fa160e3", size = 27304, upload-time = "2025-05-21T12:46:12.502Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/95/e1/df13fe3ddbbea43567e07437f097863b20c99318ae1f58a0fe389f763738/rpds_py-0.25.1-cp311-cp311-macosx_10_12_x86_64.whl", hash = "sha256:5f048bbf18b1f9120685c6d6bb70cc1a52c8cc11bdd04e643d28d3be0baf666d", size = 373341, upload-time = "2025-05-21T12:43:02.978Z" },
{ url = "https://files.pythonhosted.org/packages/7a/58/deef4d30fcbcbfef3b6d82d17c64490d5c94585a2310544ce8e2d3024f83/rpds_py-0.25.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4fbb0dbba559959fcb5d0735a0f87cdbca9e95dac87982e9b95c0f8f7ad10255", size = 359111, upload-time = "2025-05-21T12:43:05.128Z" },
{ url = "https://files.pythonhosted.org/packages/bb/7e/39f1f4431b03e96ebaf159e29a0f82a77259d8f38b2dd474721eb3a8ac9b/rpds_py-0.25.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d4ca54b9cf9d80b4016a67a0193ebe0bcf29f6b0a96f09db942087e294d3d4c2", size = 386112, upload-time = "2025-05-21T12:43:07.13Z" },
{ url = "https://files.pythonhosted.org/packages/db/e7/847068a48d63aec2ae695a1646089620b3b03f8ccf9f02c122ebaf778f3c/rpds_py-0.25.1-cp311-cp311-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:1ee3e26eb83d39b886d2cb6e06ea701bba82ef30a0de044d34626ede51ec98b0", size = 400362, upload-time = "2025-05-21T12:43:08.693Z" },
{ url = "https://files.pythonhosted.org/packages/3b/3d/9441d5db4343d0cee759a7ab4d67420a476cebb032081763de934719727b/rpds_py-0.25.1-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:89706d0683c73a26f76a5315d893c051324d771196ae8b13e6ffa1ffaf5e574f", size = 522214, upload-time = "2025-05-21T12:43:10.694Z" },
{ url = "https://files.pythonhosted.org/packages/a2/ec/2cc5b30d95f9f1a432c79c7a2f65d85e52812a8f6cbf8768724571710786/rpds_py-0.25.1-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:c2013ee878c76269c7b557a9a9c042335d732e89d482606990b70a839635feb7", size = 411491, upload-time = "2025-05-21T12:43:12.739Z" },
{ url = "https://files.pythonhosted.org/packages/dc/6c/44695c1f035077a017dd472b6a3253553780837af2fac9b6ac25f6a5cb4d/rpds_py-0.25.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:45e484db65e5380804afbec784522de84fa95e6bb92ef1bd3325d33d13efaebd", size = 386978, upload-time = "2025-05-21T12:43:14.25Z" },
{ url = "https://files.pythonhosted.org/packages/b1/74/b4357090bb1096db5392157b4e7ed8bb2417dc7799200fcbaee633a032c9/rpds_py-0.25.1-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:48d64155d02127c249695abb87d39f0faf410733428d499867606be138161d65", size = 420662, upload-time = "2025-05-21T12:43:15.8Z" },
{ url = "https://files.pythonhosted.org/packages/26/dd/8cadbebf47b96e59dfe8b35868e5c38a42272699324e95ed522da09d3a40/rpds_py-0.25.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:048893e902132fd6548a2e661fb38bf4896a89eea95ac5816cf443524a85556f", size = 563385, upload-time = "2025-05-21T12:43:17.78Z" },
{ url = "https://files.pythonhosted.org/packages/c3/ea/92960bb7f0e7a57a5ab233662f12152085c7dc0d5468534c65991a3d48c9/rpds_py-0.25.1-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:0317177b1e8691ab5879f4f33f4b6dc55ad3b344399e23df2e499de7b10a548d", size = 592047, upload-time = "2025-05-21T12:43:19.457Z" },
{ url = "https://files.pythonhosted.org/packages/61/ad/71aabc93df0d05dabcb4b0c749277881f8e74548582d96aa1bf24379493a/rpds_py-0.25.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:bffcf57826d77a4151962bf1701374e0fc87f536e56ec46f1abdd6a903354042", size = 557863, upload-time = "2025-05-21T12:43:21.69Z" },
{ url = "https://files.pythonhosted.org/packages/93/0f/89df0067c41f122b90b76f3660028a466eb287cbe38efec3ea70e637ca78/rpds_py-0.25.1-cp311-cp311-win32.whl", hash = "sha256:cda776f1967cb304816173b30994faaf2fd5bcb37e73118a47964a02c348e1bc", size = 219627, upload-time = "2025-05-21T12:43:23.311Z" },
{ url = "https://files.pythonhosted.org/packages/7c/8d/93b1a4c1baa903d0229374d9e7aa3466d751f1d65e268c52e6039c6e338e/rpds_py-0.25.1-cp311-cp311-win_amd64.whl", hash = "sha256:dc3c1ff0abc91444cd20ec643d0f805df9a3661fcacf9c95000329f3ddf268a4", size = 231603, upload-time = "2025-05-21T12:43:25.145Z" },
{ url = "https://files.pythonhosted.org/packages/cb/11/392605e5247bead2f23e6888e77229fbd714ac241ebbebb39a1e822c8815/rpds_py-0.25.1-cp311-cp311-win_arm64.whl", hash = "sha256:5a3ddb74b0985c4387719fc536faced33cadf2172769540c62e2a94b7b9be1c4", size = 223967, upload-time = "2025-05-21T12:43:26.566Z" },
{ url = "https://files.pythonhosted.org/packages/7f/81/28ab0408391b1dc57393653b6a0cf2014cc282cc2909e4615e63e58262be/rpds_py-0.25.1-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:b5ffe453cde61f73fea9430223c81d29e2fbf412a6073951102146c84e19e34c", size = 364647, upload-time = "2025-05-21T12:43:28.559Z" },
{ url = "https://files.pythonhosted.org/packages/2c/9a/7797f04cad0d5e56310e1238434f71fc6939d0bc517192a18bb99a72a95f/rpds_py-0.25.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:115874ae5e2fdcfc16b2aedc95b5eef4aebe91b28e7e21951eda8a5dc0d3461b", size = 350454, upload-time = "2025-05-21T12:43:30.615Z" },
{ url = "https://files.pythonhosted.org/packages/69/3c/93d2ef941b04898011e5d6eaa56a1acf46a3b4c9f4b3ad1bbcbafa0bee1f/rpds_py-0.25.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a714bf6e5e81b0e570d01f56e0c89c6375101b8463999ead3a93a5d2a4af91fa", size = 389665, upload-time = "2025-05-21T12:43:32.629Z" },
{ url = "https://files.pythonhosted.org/packages/c1/57/ad0e31e928751dde8903a11102559628d24173428a0f85e25e187defb2c1/rpds_py-0.25.1-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:35634369325906bcd01577da4c19e3b9541a15e99f31e91a02d010816b49bfda", size = 403873, upload-time = "2025-05-21T12:43:34.576Z" },
{ url = "https://files.pythonhosted.org/packages/16/ad/c0c652fa9bba778b4f54980a02962748479dc09632e1fd34e5282cf2556c/rpds_py-0.25.1-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d4cb2b3ddc16710548801c6fcc0cfcdeeff9dafbc983f77265877793f2660309", size = 525866, upload-time = "2025-05-21T12:43:36.123Z" },
{ url = "https://files.pythonhosted.org/packages/2a/39/3e1839bc527e6fcf48d5fec4770070f872cdee6c6fbc9b259932f4e88a38/rpds_py-0.25.1-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:9ceca1cf097ed77e1a51f1dbc8d174d10cb5931c188a4505ff9f3e119dfe519b", size = 416886, upload-time = "2025-05-21T12:43:38.034Z" },
{ url = "https://files.pythonhosted.org/packages/7a/95/dd6b91cd4560da41df9d7030a038298a67d24f8ca38e150562644c829c48/rpds_py-0.25.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2c2cd1a4b0c2b8c5e31ffff50d09f39906fe351389ba143c195566056c13a7ea", size = 390666, upload-time = "2025-05-21T12:43:40.065Z" },
{ url = "https://files.pythonhosted.org/packages/64/48/1be88a820e7494ce0a15c2d390ccb7c52212370badabf128e6a7bb4cb802/rpds_py-0.25.1-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:1de336a4b164c9188cb23f3703adb74a7623ab32d20090d0e9bf499a2203ad65", size = 425109, upload-time = "2025-05-21T12:43:42.263Z" },
{ url = "https://files.pythonhosted.org/packages/cf/07/3e2a17927ef6d7720b9949ec1b37d1e963b829ad0387f7af18d923d5cfa5/rpds_py-0.25.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:9fca84a15333e925dd59ce01da0ffe2ffe0d6e5d29a9eeba2148916d1824948c", size = 567244, upload-time = "2025-05-21T12:43:43.846Z" },
{ url = "https://files.pythonhosted.org/packages/d2/e5/76cf010998deccc4f95305d827847e2eae9c568099c06b405cf96384762b/rpds_py-0.25.1-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:88ec04afe0c59fa64e2f6ea0dd9657e04fc83e38de90f6de201954b4d4eb59bd", size = 596023, upload-time = "2025-05-21T12:43:45.932Z" },
{ url = "https://files.pythonhosted.org/packages/52/9a/df55efd84403736ba37a5a6377b70aad0fd1cb469a9109ee8a1e21299a1c/rpds_py-0.25.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a8bd2f19e312ce3e1d2c635618e8a8d8132892bb746a7cf74780a489f0f6cdcb", size = 561634, upload-time = "2025-05-21T12:43:48.263Z" },
{ url = "https://files.pythonhosted.org/packages/ab/aa/dc3620dd8db84454aaf9374bd318f1aa02578bba5e567f5bf6b79492aca4/rpds_py-0.25.1-cp312-cp312-win32.whl", hash = "sha256:e5e2f7280d8d0d3ef06f3ec1b4fd598d386cc6f0721e54f09109a8132182fbfe", size = 222713, upload-time = "2025-05-21T12:43:49.897Z" },
{ url = "https://files.pythonhosted.org/packages/a3/7f/7cef485269a50ed5b4e9bae145f512d2a111ca638ae70cc101f661b4defd/rpds_py-0.25.1-cp312-cp312-win_amd64.whl", hash = "sha256:db58483f71c5db67d643857404da360dce3573031586034b7d59f245144cc192", size = 235280, upload-time = "2025-05-21T12:43:51.893Z" },
{ url = "https://files.pythonhosted.org/packages/99/f2/c2d64f6564f32af913bf5f3f7ae41c7c263c5ae4c4e8f1a17af8af66cd46/rpds_py-0.25.1-cp312-cp312-win_arm64.whl", hash = "sha256:6d50841c425d16faf3206ddbba44c21aa3310a0cebc3c1cdfc3e3f4f9f6f5728", size = 225399, upload-time = "2025-05-21T12:43:53.351Z" },
{ url = "https://files.pythonhosted.org/packages/2b/da/323848a2b62abe6a0fec16ebe199dc6889c5d0a332458da8985b2980dffe/rpds_py-0.25.1-cp313-cp313-macosx_10_12_x86_64.whl", hash = "sha256:659d87430a8c8c704d52d094f5ba6fa72ef13b4d385b7e542a08fc240cb4a559", size = 364498, upload-time = "2025-05-21T12:43:54.841Z" },
{ url = "https://files.pythonhosted.org/packages/1f/b4/4d3820f731c80fd0cd823b3e95b9963fec681ae45ba35b5281a42382c67d/rpds_py-0.25.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:68f6f060f0bbdfb0245267da014d3a6da9be127fe3e8cc4a68c6f833f8a23bb1", size = 350083, upload-time = "2025-05-21T12:43:56.428Z" },
{ url = "https://files.pythonhosted.org/packages/d5/b1/3a8ee1c9d480e8493619a437dec685d005f706b69253286f50f498cbdbcf/rpds_py-0.25.1-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:083a9513a33e0b92cf6e7a6366036c6bb43ea595332c1ab5c8ae329e4bcc0a9c", size = 389023, upload-time = "2025-05-21T12:43:57.995Z" },
{ url = "https://files.pythonhosted.org/packages/3b/31/17293edcfc934dc62c3bf74a0cb449ecd549531f956b72287203e6880b87/rpds_py-0.25.1-cp313-cp313-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:816568614ecb22b18a010c7a12559c19f6fe993526af88e95a76d5a60b8b75fb", size = 403283, upload-time = "2025-05-21T12:43:59.546Z" },
{ url = "https://files.pythonhosted.org/packages/d1/ca/e0f0bc1a75a8925024f343258c8ecbd8828f8997ea2ac71e02f67b6f5299/rpds_py-0.25.1-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3c6564c0947a7f52e4792983f8e6cf9bac140438ebf81f527a21d944f2fd0a40", size = 524634, upload-time = "2025-05-21T12:44:01.087Z" },
{ url = "https://files.pythonhosted.org/packages/3e/03/5d0be919037178fff33a6672ffc0afa04ea1cfcb61afd4119d1b5280ff0f/rpds_py-0.25.1-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:5c4a128527fe415d73cf1f70a9a688d06130d5810be69f3b553bf7b45e8acf79", size = 416233, upload-time = "2025-05-21T12:44:02.604Z" },
{ url = "https://files.pythonhosted.org/packages/05/7c/8abb70f9017a231c6c961a8941403ed6557664c0913e1bf413cbdc039e75/rpds_py-0.25.1-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a49e1d7a4978ed554f095430b89ecc23f42014a50ac385eb0c4d163ce213c325", size = 390375, upload-time = "2025-05-21T12:44:04.162Z" },
{ url = "https://files.pythonhosted.org/packages/7a/ac/a87f339f0e066b9535074a9f403b9313fd3892d4a164d5d5f5875ac9f29f/rpds_py-0.25.1-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:d74ec9bc0e2feb81d3f16946b005748119c0f52a153f6db6a29e8cd68636f295", size = 424537, upload-time = "2025-05-21T12:44:06.175Z" },
{ url = "https://files.pythonhosted.org/packages/1f/8f/8d5c1567eaf8c8afe98a838dd24de5013ce6e8f53a01bd47fe8bb06b5533/rpds_py-0.25.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:3af5b4cc10fa41e5bc64e5c198a1b2d2864337f8fcbb9a67e747e34002ce812b", size = 566425, upload-time = "2025-05-21T12:44:08.242Z" },
{ url = "https://files.pythonhosted.org/packages/95/33/03016a6be5663b389c8ab0bbbcca68d9e96af14faeff0a04affcb587e776/rpds_py-0.25.1-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:79dc317a5f1c51fd9c6a0c4f48209c6b8526d0524a6904fc1076476e79b00f98", size = 595197, upload-time = "2025-05-21T12:44:10.449Z" },
{ url = "https://files.pythonhosted.org/packages/33/8d/da9f4d3e208c82fda311bff0cf0a19579afceb77cf456e46c559a1c075ba/rpds_py-0.25.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:1521031351865e0181bc585147624d66b3b00a84109b57fcb7a779c3ec3772cd", size = 561244, upload-time = "2025-05-21T12:44:12.387Z" },
{ url = "https://files.pythonhosted.org/packages/e2/b3/39d5dcf7c5f742ecd6dbc88f6f84ae54184b92f5f387a4053be2107b17f1/rpds_py-0.25.1-cp313-cp313-win32.whl", hash = "sha256:5d473be2b13600b93a5675d78f59e63b51b1ba2d0476893415dfbb5477e65b31", size = 222254, upload-time = "2025-05-21T12:44:14.261Z" },
{ url = "https://files.pythonhosted.org/packages/5f/19/2d6772c8eeb8302c5f834e6d0dfd83935a884e7c5ce16340c7eaf89ce925/rpds_py-0.25.1-cp313-cp313-win_amd64.whl", hash = "sha256:a7b74e92a3b212390bdce1d93da9f6488c3878c1d434c5e751cbc202c5e09500", size = 234741, upload-time = "2025-05-21T12:44:16.236Z" },
{ url = "https://files.pythonhosted.org/packages/5b/5a/145ada26cfaf86018d0eb304fe55eafdd4f0b6b84530246bb4a7c4fb5c4b/rpds_py-0.25.1-cp313-cp313-win_arm64.whl", hash = "sha256:dd326a81afe332ede08eb39ab75b301d5676802cdffd3a8f287a5f0b694dc3f5", size = 224830, upload-time = "2025-05-21T12:44:17.749Z" },
{ url = "https://files.pythonhosted.org/packages/4b/ca/d435844829c384fd2c22754ff65889c5c556a675d2ed9eb0e148435c6690/rpds_py-0.25.1-cp313-cp313t-macosx_10_12_x86_64.whl", hash = "sha256:a58d1ed49a94d4183483a3ce0af22f20318d4a1434acee255d683ad90bf78129", size = 359668, upload-time = "2025-05-21T12:44:19.322Z" },
{ url = "https://files.pythonhosted.org/packages/1f/01/b056f21db3a09f89410d493d2f6614d87bb162499f98b649d1dbd2a81988/rpds_py-0.25.1-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:f251bf23deb8332823aef1da169d5d89fa84c89f67bdfb566c49dea1fccfd50d", size = 345649, upload-time = "2025-05-21T12:44:20.962Z" },
{ url = "https://files.pythonhosted.org/packages/e0/0f/e0d00dc991e3d40e03ca36383b44995126c36b3eafa0ccbbd19664709c88/rpds_py-0.25.1-cp313-cp313t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:8dbd586bfa270c1103ece2109314dd423df1fa3d9719928b5d09e4840cec0d72", size = 384776, upload-time = "2025-05-21T12:44:22.516Z" },
{ url = "https://files.pythonhosted.org/packages/9f/a2/59374837f105f2ca79bde3c3cd1065b2f8c01678900924949f6392eab66d/rpds_py-0.25.1-cp313-cp313t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:6d273f136e912aa101a9274c3145dcbddbe4bac560e77e6d5b3c9f6e0ed06d34", size = 395131, upload-time = "2025-05-21T12:44:24.147Z" },
{ url = "https://files.pythonhosted.org/packages/9c/dc/48e8d84887627a0fe0bac53f0b4631e90976fd5d35fff8be66b8e4f3916b/rpds_py-0.25.1-cp313-cp313t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:666fa7b1bd0a3810a7f18f6d3a25ccd8866291fbbc3c9b912b917a6715874bb9", size = 520942, upload-time = "2025-05-21T12:44:25.915Z" },
{ url = "https://files.pythonhosted.org/packages/7c/f5/ee056966aeae401913d37befeeab57a4a43a4f00099e0a20297f17b8f00c/rpds_py-0.25.1-cp313-cp313t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:921954d7fbf3fccc7de8f717799304b14b6d9a45bbeec5a8d7408ccbf531faf5", size = 411330, upload-time = "2025-05-21T12:44:27.638Z" },
{ url = "https://files.pythonhosted.org/packages/ab/74/b2cffb46a097cefe5d17f94ede7a174184b9d158a0aeb195f39f2c0361e8/rpds_py-0.25.1-cp313-cp313t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f3d86373ff19ca0441ebeb696ef64cb58b8b5cbacffcda5a0ec2f3911732a194", size = 387339, upload-time = "2025-05-21T12:44:29.292Z" },
{ url = "https://files.pythonhosted.org/packages/7f/9a/0ff0b375dcb5161c2b7054e7d0b7575f1680127505945f5cabaac890bc07/rpds_py-0.25.1-cp313-cp313t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:c8980cde3bb8575e7c956a530f2c217c1d6aac453474bf3ea0f9c89868b531b6", size = 418077, upload-time = "2025-05-21T12:44:30.877Z" },
{ url = "https://files.pythonhosted.org/packages/0d/a1/fda629bf20d6b698ae84c7c840cfb0e9e4200f664fc96e1f456f00e4ad6e/rpds_py-0.25.1-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:8eb8c84ecea987a2523e057c0d950bcb3f789696c0499290b8d7b3107a719d78", size = 562441, upload-time = "2025-05-21T12:44:32.541Z" },
{ url = "https://files.pythonhosted.org/packages/20/15/ce4b5257f654132f326f4acd87268e1006cc071e2c59794c5bdf4bebbb51/rpds_py-0.25.1-cp313-cp313t-musllinux_1_2_i686.whl", hash = "sha256:e43a005671a9ed5a650f3bc39e4dbccd6d4326b24fb5ea8be5f3a43a6f576c72", size = 590750, upload-time = "2025-05-21T12:44:34.557Z" },
{ url = "https://files.pythonhosted.org/packages/fb/ab/e04bf58a8d375aeedb5268edcc835c6a660ebf79d4384d8e0889439448b0/rpds_py-0.25.1-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:58f77c60956501a4a627749a6dcb78dac522f249dd96b5c9f1c6af29bfacfb66", size = 558891, upload-time = "2025-05-21T12:44:37.358Z" },
{ url = "https://files.pythonhosted.org/packages/90/82/cb8c6028a6ef6cd2b7991e2e4ced01c854b6236ecf51e81b64b569c43d73/rpds_py-0.25.1-cp313-cp313t-win32.whl", hash = "sha256:2cb9e5b5e26fc02c8a4345048cd9998c2aca7c2712bd1b36da0c72ee969a3523", size = 218718, upload-time = "2025-05-21T12:44:38.969Z" },
{ url = "https://files.pythonhosted.org/packages/b6/97/5a4b59697111c89477d20ba8a44df9ca16b41e737fa569d5ae8bff99e650/rpds_py-0.25.1-cp313-cp313t-win_amd64.whl", hash = "sha256:401ca1c4a20cc0510d3435d89c069fe0a9ae2ee6495135ac46bdd49ec0495763", size = 232218, upload-time = "2025-05-21T12:44:40.512Z" },
{ url = "https://files.pythonhosted.org/packages/49/74/48f3df0715a585cbf5d34919c9c757a4c92c1a9eba059f2d334e72471f70/rpds_py-0.25.1-pp311-pypy311_pp73-macosx_10_12_x86_64.whl", hash = "sha256:ee86d81551ec68a5c25373c5643d343150cc54672b5e9a0cafc93c1870a53954", size = 374208, upload-time = "2025-05-21T12:45:26.306Z" },
{ url = "https://files.pythonhosted.org/packages/55/b0/9b01bb11ce01ec03d05e627249cc2c06039d6aa24ea5a22a39c312167c10/rpds_py-0.25.1-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:89c24300cd4a8e4a51e55c31a8ff3918e6651b241ee8876a42cc2b2a078533ba", size = 359262, upload-time = "2025-05-21T12:45:28.322Z" },
{ url = "https://files.pythonhosted.org/packages/a9/eb/5395621618f723ebd5116c53282052943a726dba111b49cd2071f785b665/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:771c16060ff4e79584dc48902a91ba79fd93eade3aa3a12d6d2a4aadaf7d542b", size = 387366, upload-time = "2025-05-21T12:45:30.42Z" },
{ url = "https://files.pythonhosted.org/packages/68/73/3d51442bdb246db619d75039a50ea1cf8b5b4ee250c3e5cd5c3af5981cd4/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:785ffacd0ee61c3e60bdfde93baa6d7c10d86f15655bd706c89da08068dc5038", size = 400759, upload-time = "2025-05-21T12:45:32.516Z" },
{ url = "https://files.pythonhosted.org/packages/b7/4c/3a32d5955d7e6cb117314597bc0f2224efc798428318b13073efe306512a/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:2a40046a529cc15cef88ac5ab589f83f739e2d332cb4d7399072242400ed68c9", size = 523128, upload-time = "2025-05-21T12:45:34.396Z" },
{ url = "https://files.pythonhosted.org/packages/be/95/1ffccd3b0bb901ae60b1dd4b1be2ab98bb4eb834cd9b15199888f5702f7b/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:85fc223d9c76cabe5d0bff82214459189720dc135db45f9f66aa7cffbf9ff6c1", size = 411597, upload-time = "2025-05-21T12:45:36.164Z" },
{ url = "https://files.pythonhosted.org/packages/ef/6d/6e6cd310180689db8b0d2de7f7d1eabf3fb013f239e156ae0d5a1a85c27f/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b0be9965f93c222fb9b4cc254235b3b2b215796c03ef5ee64f995b1b69af0762", size = 388053, upload-time = "2025-05-21T12:45:38.45Z" },
{ url = "https://files.pythonhosted.org/packages/4a/87/ec4186b1fe6365ced6fa470960e68fc7804bafbe7c0cf5a36237aa240efa/rpds_py-0.25.1-pp311-pypy311_pp73-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:8378fa4a940f3fb509c081e06cb7f7f2adae8cf46ef258b0e0ed7519facd573e", size = 421821, upload-time = "2025-05-21T12:45:40.732Z" },
{ url = "https://files.pythonhosted.org/packages/7a/60/84f821f6bf4e0e710acc5039d91f8f594fae0d93fc368704920d8971680d/rpds_py-0.25.1-pp311-pypy311_pp73-musllinux_1_2_aarch64.whl", hash = "sha256:33358883a4490287e67a2c391dfaea4d9359860281db3292b6886bf0be3d8692", size = 564534, upload-time = "2025-05-21T12:45:42.672Z" },
{ url = "https://files.pythonhosted.org/packages/41/3a/bc654eb15d3b38f9330fe0f545016ba154d89cdabc6177b0295910cd0ebe/rpds_py-0.25.1-pp311-pypy311_pp73-musllinux_1_2_i686.whl", hash = "sha256:1d1fadd539298e70cac2f2cb36f5b8a65f742b9b9f1014dd4ea1f7785e2470bf", size = 592674, upload-time = "2025-05-21T12:45:44.533Z" },
{ url = "https://files.pythonhosted.org/packages/2e/ba/31239736f29e4dfc7a58a45955c5db852864c306131fd6320aea214d5437/rpds_py-0.25.1-pp311-pypy311_pp73-musllinux_1_2_x86_64.whl", hash = "sha256:9a46c2fb2545e21181445515960006e85d22025bd2fe6db23e76daec6eb689fe", size = 558781, upload-time = "2025-05-21T12:45:46.281Z" },
]
[[package]]
name = "ruff"
version = "0.11.13"
@@ -873,6 +1036,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" },
]
[[package]]
name = "six"
version = "1.17.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
]
[[package]]
name = "sniffio"
version = "1.3.1"
@@ -989,7 +1161,7 @@ wheels = [
[[package]]
name = "typer"
version = "0.15.3"
version = "0.16.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
@@ -997,9 +1169,9 @@ dependencies = [
{ name = "shellingham" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/98/1a/5f36851f439884bcfe8539f6a20ff7516e7b60f319bbaf69a90dc35cc2eb/typer-0.15.3.tar.gz", hash = "sha256:818873625d0569653438316567861899f7e9972f2e6e0c16dab608345ced713c", size = 101641, upload-time = "2025-04-28T21:40:59.204Z" }
sdist = { url = "https://files.pythonhosted.org/packages/c5/8c/7d682431efca5fd290017663ea4588bf6f2c6aad085c7f108c5dbc316e70/typer-0.16.0.tar.gz", hash = "sha256:af377ffaee1dbe37ae9440cb4e8f11686ea5ce4e9bae01b84ae7c63b87f1dd3b", size = 102625, upload-time = "2025-05-26T14:30:31.824Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/48/20/9d953de6f4367163d23ec823200eb3ecb0050a2609691e512c8b95827a9b/typer-0.15.3-py3-none-any.whl", hash = "sha256:c86a65ad77ca531f03de08d1b9cb67cd09ad02ddddf4b34745b5008f43b239bd", size = 45253, upload-time = "2025-04-28T21:40:56.269Z" },
{ url = "https://files.pythonhosted.org/packages/76/42/3efaf858001d2c2913de7f354563e3a3a2f0decae3efe98427125a8f441e/typer-0.16.0-py3-none-any.whl", hash = "sha256:1f79bed11d4d02d4310e3c1b7ba594183bcedb0ac73b27a9e5f28f6fb5b98855", size = 46317, upload-time = "2025-05-26T14:30:30.523Z" },
]
[[package]]
@@ -1023,6 +1195,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/31/08/aa4fdfb71f7de5176385bd9e90852eaf6b5d622735020ad600f2bab54385/typing_inspection-0.4.0-py3-none-any.whl", hash = "sha256:50e72559fcd2a6367a19f7a7e610e6afcb9fac940c650290eed893d61386832f", size = 14125, upload-time = "2025-02-25T17:27:57.754Z" },
]
[[package]]
name = "tzdata"
version = "2025.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/95/32/1a225d6164441be760d75c2c42e2780dc0873fe382da3e98a2e1e48361e5/tzdata-2025.2.tar.gz", hash = "sha256:b60a638fcc0daffadf82fe0f57e53d06bdec2f36c4df66280ae79bce6bd6f2b9", size = 196380, upload-time = "2025-03-23T13:54:43.652Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5c/23/c7abc0ca0a1526a0774eca151daeb8de62ec457e77262b66b359c3c7679e/tzdata-2025.2-py2.py3-none-any.whl", hash = "sha256:1a403fada01ff9221ca8044d701868fa132215d84beb92242d9acd2147f667a8", size = 347839, upload-time = "2025-03-23T13:54:41.845Z" },
]
[[package]]
name = "uvicorn"
version = "0.34.2"