feat: Add tools for managing server scripts, client scripts, translations, assignment rules, user permissions, webhooks, API keys, and workflows

- Implemented server and client script management tools in `frappe_mcp/tools/scripts.py`
- Added translation and user permission management tools in `frappe_mcp/tools/translations.py`
- Created user and role management tools in `frappe_mcp/tools/users.py`
- Developed webhook and API key management tools in `frappe_mcp/tools/webhooks.py`
- Introduced workflow management tools in `frappe_mcp/tools/workflow_tools.py`
- Added `pyproject.toml` for project metadata and dependencies
This commit is contained in:
MOHAN 2026-04-21 20:26:45 +05:30
commit 2ee93048e1
110 changed files with 6813 additions and 0 deletions

47
.env Normal file
View File

@ -0,0 +1,47 @@
# Frappe instance URL (your VPS / Docker host)
# If using a domain: https://erp.yourdomain.com
# If using Docker with port mapping: http://YOUR_VPS_IP:8000
FRAPPE_URL=http://147.93.40.215:10009/
# API credentials — generate in Frappe: Settings > My Account > API Access
FRAPPE_API_KEY=2650fa15dc9393f
FRAPPE_API_SECRET=766ad5af8577685
# For multi-site Docker setups (optional) — the site name e.g. "site1.localhost"
FRAPPE_SITE_NAME=
# Set to true to block all write/delete operations (safe read-only mode)
READ_ONLY_MODE=false
# HTTP timeout in seconds for API calls
REQUEST_TIMEOUT=30
ENABLED_MODULES=foundation,document_inspect,analytics
# ── Module on/off switches ───────────────────────────────────────────────────
# Uncomment a line to disable that module (all others stay ON)
# MODULE_FOUNDATION=false
# MODULE_DOCTYPES=false
# MODULE_DOCUMENTS=false
# MODULE_DOCUMENT_INSPECT=false
# MODULE_DOCUMENT_LIFECYCLE=false
# MODULE_CUSTOM_FIELDS=false
# MODULE_SCRIPTS=false
# MODULE_WORKFLOW_TOOLS=false
# MODULE_ANALYTICS=false
# MODULE_REPORTS=false
# MODULE_PRINT_FORMATS=false
# MODULE_BULK_OPS=false
# MODULE_BUSINESS_ACTIONS=false
# MODULE_USERS=false
# MODULE_ADMIN=false
# MODULE_PROPERTY_SETTERS=false
# MODULE_NAMING_SERIES=false
# MODULE_FILES=false
# MODULE_ACTIVITY=false
# MODULE_DASHBOARDS=false
# MODULE_WEBHOOKS=false
# MODULE_EMAIL_TEMPLATES=false
# MODULE_TRANSLATIONS=false
# MODULE_SCHEDULER=false
# MODULE_GOVERNANCE=false

72
.env.example Normal file
View File

@ -0,0 +1,72 @@
# ── Frappe Connection ────────────────────────────────────────────────────────
# URL of your Frappe Docker instance (VPS IP or domain)
FRAPPE_URL=http://YOUR_VPS_IP:8000
# API credentials — generate in Frappe: Settings > My Profile > API Access
FRAPPE_API_KEY=your_api_key_here
FRAPPE_API_SECRET=your_api_secret_here
# For multi-site Docker setups — the site name e.g. "site1.localhost"
FRAPPE_SITE_NAME=
# ── Safety ───────────────────────────────────────────────────────────────────
# Set to true to block ALL write/delete operations (read-only audit mode)
READ_ONLY_MODE=false
# HTTP timeout for API calls (seconds)
REQUEST_TIMEOUT=30
# ── Module Activation ────────────────────────────────────────────────────────
# Option A: Enable ONLY specific modules (comma-separated, all others disabled)
# ENABLED_MODULES=documents,doctypes,users,custom_fields
# Option B: Disable specific modules (all others stay ON)
# MODULE_SCHEDULER=false
# MODULE_BULK_OPS=false
# MODULE_TRANSLATIONS=false
# MODULE_WEBHOOKS=false
# Available module keys (run 'frappe-mcp --list-modules' for live status):
#
# LEVEL 1 — Foundation (always recommended ON)
# foundation — ping, session_info, doctype_meta, permissions
#
# LEVEL 2 — Read
# doctypes — DocType CRUD
# documents — Document CRUD (submit, cancel, delete)
# document_inspect — search, children, linked docs, timeline, count
#
# LEVEL 3-4 — Write + Lifecycle
# document_lifecycle — child rows, rename, amend, duplicate, status
# custom_fields — Custom Fields
# scripts — Server Scripts + Client Scripts
#
# LEVEL 5 — Workflow
# workflow_tools — Workflows, transitions, approvals
#
# LEVEL 6 — Reporting
# analytics — Aggregate, dashboard data, PDF render
# reports — Query/Script Report CRUD
# print_formats — Print Format templates
#
# LEVEL 7 — Bulk
# bulk_ops — Bulk create/update/delete/submit/cancel/assign/tag
#
# LEVEL 8 — Business Actions (ERPNext)
# business_actions — Sales, Buying, Stock, HR, Support, Projects shortcuts
#
# LEVEL 9 — Admin / System
# users — Users, Roles, Permissions
# admin — Cache, System Settings, SQL, Workflows
# property_setters — Property Setters + Customize Form
# naming_series — Naming Series
# files — File Manager
# activity — Comments, Tags, Assignments, Error Logs
# dashboards — Dashboards, Charts, Workspaces
# webhooks — Webhooks + API Keys
# email_templates — Email Templates + Notifications
# translations — Translations, Assignment Rules, User Permissions
# scheduler — Scheduled Jobs + Background Jobs
#
# LEVEL 10 — Governance (enable for autonomous AI agents)
# governance — Dry run, validate, risk score, audit log, rollback

16
.mcp.json Normal file
View File

@ -0,0 +1,16 @@
{
"mcpServers": {
"frappe": {
"type": "stdio",
"command": "python",
"args": ["-m", "frappe_mcp.server"],
"cwd": "D:\\2026\\Frappe-MCP",
"env": {
"FRAPPE_URL": "http://147.93.40.215:10009",
"FRAPPE_API_KEY": "2650fa15dc9393f",
"FRAPPE_API_SECRET": "766ad5af8577685"
}
}
}
}

347
DINE360_ARCHITECTURE.md Normal file
View File

@ -0,0 +1,347 @@
# Dine360 Architecture Plan
> Generated: 2026-04-20 | Platform: Frappe 15.105.0 + ERPNext 15.104.3
> Instance: http://147.93.40.215:10009/
---
## 1. Platform Foundation
| Layer | Version | Notes |
|-------|---------|-------|
| Frappe Framework | 15.105.0 | Core DocType engine, workflow, notifications, scheduler |
| ERPNext | 15.104.3 | CRM, Accounts, HR modules available for reuse |
| Custom Module | Dine360 | Registered under `frappe` app |
---
## 2. Standard ERPNext DocTypes — Reuse Opportunities
These exist out of the box and can be linked from Dine360 DocTypes rather than recreating:
| ERPNext DocType | Dine360 Purpose | How Reused |
|-----------------|-----------------|------------|
| **Customer** | Advertiser billing account | Link from Advertiser → Customer for sales invoices |
| **Supplier** | Restaurant Partner payout | Link from Restaurant Partner → Supplier for purchase invoices |
| **Sales Invoice** | Advertiser billing | Create against Customer when campaign runs |
| **Purchase Invoice** | Restaurant payout | Create against Supplier on settlement |
| **Payment Entry** | Record actual payments | Link from Restaurant Settlement after payout |
| **Address** | Branch physical location | Link from Restaurant Branch → Address |
| **Contact** | Partner/advertiser contacts | Standard Contact linked to Restaurant Partner/Advertiser |
| **Employee** | Internal operations staff | Assign campaigns, escalations |
| **User** | Portal access | All 8 Dine360 roles assigned to Users |
| **File** | Creative asset storage | Ad Creative links to File for media uploads |
| **Communication** | Email/SMS log | Auto-logged on notification send |
| **Activity Log** | Audit trail | Auto-logged on all document changes |
| **ToDo** | Onboarding task reminders | Auto-created from Onboarding Checklist workflow |
---
## 3. Custom Dine360 DocTypes (19 Total)
### 3a. Child Tables (4)
| DocType | Parent | Purpose |
|---------|--------|---------|
| **Screen Group Member** | Screen Group | Lists Screen Devices belonging to a group |
| **Campaign Creative Item** | Ad Campaign | Creative assets (video/image) with duration per campaign |
| **Campaign Target Item** | Ad Campaign | Target screen groups and time slots per campaign |
| **Settlement Ledger Item** | Restaurant Settlement | Per-restaurant line items in a batch settlement |
### 3b. Master / Configuration DocTypes (6)
| DocType | Submittable | Key Fields | Autoname |
|---------|-------------|------------|----------|
| **Restaurant Partner** | No | partner_name, contact_email, status (Active/Inactive/Suspended), linked_supplier | `field:partner_name` |
| **Advertiser** | No | advertiser_name, contact_email, industry, linked_customer | `field:advertiser_name` |
| **Restaurant Branch** | No | branch_name, restaurant_partner (Link), city, address, is_active | `BRANCH-.#####` |
| **Ad Creative** | No | creative_name, advertiser (Link), media_type (Video/Image/HTML), file_url, duration_seconds, status (Draft/Approved/Rejected) | `CRTV-.YYYY.-.#####` |
| **Screen Device** | No | device_name, restaurant_branch (Link), screen_group (Link), device_serial, status (Active/Inactive/Maintenance), last_heartbeat | `SCRN-.#####` |
| **Screen Group** | No | group_name, description, screen_group_members (child table) | `field:group_name` |
### 3c. Transaction DocTypes (9)
| DocType | Submittable | Key Fields | Autoname |
|---------|-------------|------------|----------|
| **Ad Campaign** | **Yes** | campaign_name, advertiser (Link), start_date, end_date, budget, status, campaign_creative_items (child), campaign_target_items (child) | `CAMP-.YYYY.-.#####` |
| **Revenue Rule** | No | rule_name, advertiser (Link), restaurant_partner (Link), revenue_split_percent, effective_from, effective_to | `RULE-.YYYY.-.#####` |
| **API Sync Log** | No | sync_type, sync_status (Success/Failed/Partial), synced_at, records_processed, error_message | `SYNC-.YYYY.-.#####` |
| **Campaign Schedule** | No | ad_campaign (Link), screen_device (Link), scheduled_date, start_time, end_time, status (Pending/Confirmed/Cancelled) | `SCHED-.YYYY.-.#####` |
| **Playback Log** | No | screen_device (Link), ad_campaign (Link), played_at, duration_played, completion_status | `PLAY-.YYYY.-.#####` |
| **Device Heartbeat** | No | screen_device (Link), heartbeat_at, ip_address, cpu_load, memory_used, status (Online/Offline/Warning) | `HB-.YYYY.-.#####` |
| **Revenue Ledger** | No | restaurant_partner (Link), ad_campaign (Link), revenue_rule (Link), period_start, period_end, gross_revenue, platform_share, partner_share, status (Draft/Calculated/Approved) | `RLGR-.YYYY.-.#####` |
| **Restaurant Settlement** | **Yes** | settlement_name, settlement_period_start, settlement_period_end, total_payout, status, settlement_ledger_items (child) | `SETL-.YYYY.-.#####` |
| **Onboarding Checklist** | No | restaurant_partner (Link), assigned_to, current_stage, lead_date, review_date, agreement_date, setup_date, go_live_date | `OB-.YYYY.-.#####` |
---
## 4. Entity Relationship Map
```
┌─────────────────┐
│ Advertiser │──────────────────────────┐
└────────┬────────┘ │
│ 1 │ 1
▼ N ▼ N
┌────────────────┐ ┌────────────────────┐
│ Ad Creative │ │ Revenue Rule │
└────────┬───────┘ └────────────────────┘
│ N │
▼ M (via child table) │
┌────────────────┐ │
│ Ad Campaign │─────────────────────────┘
└────┬───────┬───┘
│ │ via Campaign Target Items
│ ▼ N
│ ┌─────────────┐ ┌──────────────────┐
│ │ Screen Group│───────│ Screen Group │
│ └─────────────┘ │ Member (child) │
│ │ └──────────────────┘
│ │ N
│ ▼ 1
│ ┌─────────────┐ ┌───────────────────┐
│ │Screen Device│───────│Restaurant Branch │
│ └──────┬──────┘ └────────┬──────────┘
│ │ │ N
│ │ 1 ▼ 1
│ ┌──────┴──────┐ ┌──────────────────────┐
│ │ Playback │ │ Restaurant Partner │
│ │ Log │ └────────────┬─────────┘
│ └─────────────┘ │ 1
│ ▼ N
│ ┌──────────────────────┐
└─────────────────────────►│ Revenue Ledger │
└──────────┬───────────┘
│ N
▼ 1 (via child)
┌──────────────────────┐
│Restaurant Settlement │
└──────────────────────┘
Device Heartbeat ──► Screen Device
Campaign Schedule ──► Ad Campaign + Screen Device
API Sync Log ──► (standalone audit log)
Onboarding Checklist ──► Restaurant Partner
```
---
## 5. Role Permission Matrix
| Role | Masters | Campaigns | Revenue | Devices | Settlement | Onboarding |
|------|---------|-----------|---------|---------|------------|------------|
| **Dine360 Admin** | Full | Full | Full | Full | Full | Full |
| **Campaign Manager** | Read | Full | Read | Read | None | None |
| **Operations Manager** | Read | Read/Write | Read | Full | Read | Full |
| **Finance Manager** | Read | Read | Full | None | Full | None |
| **Support Executive** | Read | Read | None | Read | None | Read/Write |
| **Restaurant Manager** | Own | Own | Own | Own | Own | Own |
| **Restaurant Viewer** | Read-Own | Read-Own | Read-Own | None | Read-Own | None |
| **Device Agent** | None | Read | None | Write-Own | None | None |
---
## 6. Workflow Map
### 6a. Campaign Approval Workflow (Ad Campaign)
```
Draft ──[Submit for Review]──► Pending Approval ──[Approve]──► Approved ──[Schedule]──► Scheduled
│ │ │
[Reject] [Reject] [Activate]
│ │ │
▼ ▼ ▼
Rejected Rejected Running ──[Complete]──► Completed
[Cancel]
Cancelled
```
| Transition | From | To | Action | Roles |
|------------|------|----|--------|-------|
| Submit for Review | Draft | Pending Approval | Submit for Review | Campaign Manager |
| Approve | Pending Approval | Approved | Approve | Dine360 Admin, Operations Manager |
| Reject | Pending Approval | Rejected | Reject | Dine360 Admin, Operations Manager |
| Schedule | Approved | Scheduled | Schedule | Campaign Manager, Operations Manager |
| Activate | Scheduled | Running | Activate | Device Agent, Operations Manager |
| Complete | Running | Completed | Complete | Device Agent, Operations Manager |
| Cancel | Running | Cancelled | Cancel | Dine360 Admin, Operations Manager |
### 6b. Restaurant Onboarding Workflow (Onboarding Checklist)
```
Lead ──[Move to Review]──► Review ──[Sign Agreement]──► Agreement ──[Begin Setup]──► Setup ──[Go Live]──► Live
```
| Transition | From | To | Action | Roles |
|------------|------|----|--------|-------|
| Move to Review | Lead | Review | Move to Review | Operations Manager, Support Executive |
| Sign Agreement | Review | Agreement | Sign Agreement | Operations Manager, Dine360 Admin |
| Begin Setup | Agreement | Setup | Begin Setup | Operations Manager |
| Go Live | Setup | Live | Go Live | Operations Manager, Dine360 Admin |
### 6c. Settlement Payout Workflow (Restaurant Settlement)
```
Draft ──[Calculate]──► Calculated ──[Approve Payout]──► Approved ──[Mark Paid]──► Paid
│ │ │
[Hold] [Hold] [Raise Dispute]
│ │ │
▼ ▼ ▼
Held Held Disputed ──[Resolve]──► Paid
```
| Transition | From | To | docstatus | Action | Roles |
|------------|------|----|-----------|--------|-------|
| Calculate | Draft | Calculated | 0→0 | Calculate | Finance Manager |
| Approve Payout | Calculated | Approved | 0→1 | Approve Payout | Finance Manager, Dine360 Admin |
| Hold | Calculated | Held | 0→0 | Hold | Finance Manager |
| Hold | Approved | Held | 1→1 | Hold | Finance Manager, Dine360 Admin |
| Mark Paid | Approved | Paid | 1→1 | Mark Paid | Finance Manager |
| Raise Dispute | Paid | Disputed | 1→1 | Raise Dispute | Restaurant Manager, Finance Manager |
| Resolve Dispute | Disputed | Paid | 1→1 | Resolve Dispute | Finance Manager, Dine360 Admin |
---
## 7. Naming Series Reference
| DocType | Series | Example |
|---------|--------|---------|
| Restaurant Branch | `BRANCH-.#####` | BRANCH-00001 |
| Ad Creative | `CRTV-.YYYY.-.#####` | CRTV-2026-00001 |
| Screen Device | `SCRN-.#####` | SCRN-00001 |
| Ad Campaign | `CAMP-.YYYY.-.#####` | CAMP-2026-00001 |
| Revenue Rule | `RULE-.YYYY.-.#####` | RULE-2026-00001 |
| API Sync Log | `SYNC-.YYYY.-.#####` | SYNC-2026-00001 |
| Campaign Schedule | `SCHED-.YYYY.-.#####` | SCHED-2026-00001 |
| Playback Log | `PLAY-.YYYY.-.#####` | PLAY-2026-00001 |
| Device Heartbeat | `HB-.YYYY.-.#####` | HB-2026-00001 |
| Revenue Ledger | `RLGR-.YYYY.-.#####` | RLGR-2026-00001 |
| Restaurant Settlement | `SETL-.YYYY.-.#####` | SETL-2026-00001 |
| Onboarding Checklist | `OB-.YYYY.-.#####` | OB-2026-00001 |
| Restaurant Partner | `field:partner_name` | Metatron Cubes |
| Advertiser | `field:advertiser_name` | Coca-Cola India |
| Screen Group | `field:group_name` | Zone A Screens |
---
## 8. Status Values Reference
| DocType | Status Field | Allowed Values |
|---------|-------------|----------------|
| Restaurant Partner | status | Active, Inactive, Suspended |
| Ad Creative | status | Draft, Approved, Rejected |
| Screen Device | status | Active, Inactive, Maintenance |
| Campaign Schedule | status | Pending, Confirmed, Cancelled |
| Playback Log | completion_status | Completed, Partial, Failed |
| Device Heartbeat | status | Online, Offline, Warning |
| Revenue Ledger | status | Draft, Calculated, Approved |
| API Sync Log | sync_status | Success, Failed, Partial |
---
## 9. Notification Map (5 Active Notifications)
| Notification | DocType | Trigger | Recipients |
|-------------|---------|---------|-----------|
| Campaign Submitted for Review | Ad Campaign | workflow_state → Pending Approval | Dine360 Admin, Operations Manager |
| Campaign Approved | Ad Campaign | workflow_state → Approved | Campaign Manager |
| Campaign Rejected | Ad Campaign | workflow_state → Rejected | Campaign Manager |
| Settlement Approved for Payout | Restaurant Settlement | workflow_state → Approved | Finance Manager |
| Campaign Ending Soon | Ad Campaign | end_date within 3 days | Campaign Manager |
---
## 10. Dashboard & Reporting Map
### 10a. Existing Dashboard Charts (6)
| Chart | Type | DocType | X-Axis | Y-Axis |
|-------|------|---------|--------|--------|
| Campaigns by Status | Donut | Ad Campaign | — | status (count) |
| Active Screen Devices | Donut | Screen Device | — | status (count) |
| Restaurant Partners by Status | Donut | Restaurant Partner | — | status (count) |
| Monthly Revenue Ledger | Bar | Revenue Ledger | period_start (monthly) | partner_share (sum) |
| Settlements Over Time | Bar | Restaurant Settlement | settlement_period_start (monthly) | total_payout (sum) |
| Playback Log Volume | Line | Playback Log | played_at (daily) | name (count) |
### 10b. Recommended Future Reports (Phase 2)
| Report | Type | Purpose |
|--------|------|---------|
| Campaign Performance Summary | Query Report | Impressions, completion rate, revenue per campaign |
| Revenue Split Breakdown | Script Report | Per-partner revenue vs platform share over period |
| Device Uptime Report | Query Report | Heartbeat-based uptime % per device/branch |
| Settlement Audit Trail | Query Report | Full ledger→settlement→payment chain |
| Advertiser Billing Statement | Script Report | Invoice-ready summary for advertiser |
| Onboarding Pipeline | Query Report | Checklist stage distribution + SLA |
---
## 11. Server-Side Automation Map (Phase 2)
| Automation | Type | Trigger | Action |
|------------|------|---------|--------|
| Revenue Split Calculator | Server Script | On Playback Log save | Calculate partner_share in Revenue Ledger |
| Device Offline Alert | Scheduled Job | Every 5 min | Flag Device Heartbeat > 10 min old as Offline |
| Campaign Auto-Complete | Scheduled Job | Daily | Set Running campaigns past end_date → Completed |
| Settlement Auto-Batch | Scheduled Job | Monthly | Batch Calculated Revenue Ledgers → new Settlement |
| Creative Auto-Expire | Scheduled Job | Daily | Flag Ad Creatives not updated in 90 days |
| Device Sync API | API Endpoint | REST POST | Accept heartbeat + playback data from devices |
---
## 12. Phase 2 Implementation Order
All 19 DocTypes, 3 workflows, 5 notifications, 6 charts, and 1 workspace are **already live**.
Phase 2 items in recommended build order:
```
Step 1 — Server Scripts (business logic)
1.1 Revenue split auto-calculation on Playback Log save
1.2 Settlement batch-creation script (aggregate Revenue Ledger → Settlement)
1.3 Campaign status auto-complete (scheduled)
1.4 Device offline detection (scheduled heartbeat check)
Step 2 — Query / Script Reports
2.1 Campaign Performance Summary (Query Report)
2.2 Revenue Split Breakdown (Script Report)
2.3 Device Uptime Report (Query Report)
2.4 Settlement Audit Trail (Query Report)
Step 3 — Print Formats
3.1 Settlement Payout Slip (Restaurant Settlement)
3.2 Advertiser Campaign Summary (Ad Campaign)
Step 4 — Portal / Role Permissions
4.1 Tighten user permissions per Role Permission Matrix (Section 5)
4.2 Configure Restaurant Manager portal view
4.3 Configure Advertiser self-service portal (if needed)
Step 5 — API Endpoints
5.1 Device heartbeat sync endpoint (POST /api/method/dine360.heartbeat)
5.2 Playback log ingestion endpoint (POST /api/method/dine360.playback)
5.3 Campaign schedule fetch endpoint (GET /api/method/dine360.schedule)
Step 6 — Integration
6.1 Link Restaurant Partner → Supplier (for Purchase Invoice on settlement)
6.2 Link Advertiser → Customer (for Sales Invoice on campaign billing)
6.3 Auto-create Payment Entry on Settlement marked Paid
```
---
## 13. Key Design Decisions & Constraints
1. **Dine360 is a custom module** on the `frappe` app — not a separate installable app. Upgrade-safe as long as customizations live in DocType/Server Script/Client Script records.
2. **Submittable DocTypes**: Only `Ad Campaign` and `Restaurant Settlement` are submittable. Revenue Ledger uses a status field (not docstatus) to allow edits through the approval cycle.
3. **Workflow docstatus rule**: Frappe forbids transitions from submitted (docstatus=1) → draft (docstatus=0). All dispute/hold states in the Settlement workflow are docstatus=1.
4. **ERPNext accounting integration** is optional but available: Settlement → Purchase Invoice → Payment Entry chain uses native ERPNext accounting. Disable if a simpler ledger-only approach is preferred.
5. **Device authentication**: Screen Devices should authenticate via API key (generate one API key per device via `frappe_generate_api_key`). The Device Agent role limits read access to only Campaign Schedule.
6. **Revenue Rule priority**: If multiple Revenue Rules match an Advertiser + Partner pair, use the most recently effective rule. Overlap validation should be enforced via server script.

BIN
Frappe-MCP.rar Normal file

Binary file not shown.

View File

@ -0,0 +1,14 @@
{
"mcpServers": {
"frappe": {
"command": "python",
"args": ["-m", "frappe_mcp.server"],
"cwd": "/path/to/frappe-mcp",
"env": {
"FRAPPE_URL": "http://147.93.40.215:10009/",
"FRAPPE_API_KEY": "2650fa15dc9393f",
"FRAPPE_API_SECRET": "766ad5af8577685"
}
}
}
}

16
docker/.env.example Normal file
View File

@ -0,0 +1,16 @@
# ERPNext version (use version-15 for latest stable)
ERPNEXT_VERSION=version-15
# Your site name — use your domain or VPS IP
# If using a domain: myerp.yourdomain.com
# If no domain: erp.localhost
SITE_NAME=erp.localhost
# Port to expose on the VPS (80 if you want standard HTTP, 8080 for non-root)
HTTP_PORT=8080
# Database root password — change this!
DB_ROOT_PASSWORD=SuperSecureDBPass123!
# Frappe admin password — you'll use this to log in at /login
ADMIN_PASSWORD=SuperSecureAdmin123!

206
docker/docker-compose.yml Normal file
View File

@ -0,0 +1,206 @@
services:
# ── Database ────────────────────────────────────────────────────────────────
db:
image: mariadb:10.6
restart: unless-stopped
environment:
MYSQL_ROOT_PASSWORD: ${DB_ROOT_PASSWORD:-changeme123}
MYSQL_DATABASE: _sys
volumes:
- db-data:/var/lib/mysql
networks:
- frappe-net
command:
- --character-set-server=utf8mb4
- --collation-server=utf8mb4_unicode_ci
- --skip-character-set-client-handshake
- --skip-innodb-read-only-compressed
# ── Redis ───────────────────────────────────────────────────────────────────
redis-cache:
image: redis:7-alpine
restart: unless-stopped
networks:
- frappe-net
volumes:
- redis-cache-data:/data
redis-queue:
image: redis:7-alpine
restart: unless-stopped
networks:
- frappe-net
volumes:
- redis-queue-data:/data
# ── Site configurator (runs once, then exits) ───────────────────────────────
configurator:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: "no"
entrypoint: ["bash", "-c"]
command:
- >
ls -1 apps > sites/apps.txt;
bench set-config -g db_host db;
bench set-config -gp db_port 3306;
bench set-config -g redis_cache "redis://redis-cache:6379";
bench set-config -g redis_queue "redis://redis-queue:6379";
bench set-config -g redis_socketio "redis://redis-queue:6379";
bench set-config -gp socketio_port 9000;
environment:
DB_HOST: db
DB_PORT: "3306"
REDIS_CACHE: redis://redis-cache:6379
REDIS_QUEUE: redis://redis-queue:6379
SOCKETIO_PORT: "9000"
volumes:
- sites:/home/frappe/frappe-bench/sites
networks:
- frappe-net
depends_on:
- db
# ── Site creator (runs once to create site + install ERPNext) ───────────────
create-site:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: "no"
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
entrypoint: ["bash", "-c"]
command:
- >
wait-for-it -t 120 db:3306;
wait-for-it -t 120 redis-cache:6379;
wait-for-it -t 120 redis-queue:6379;
export start=`date +%s`;
until [[ -n `grep -hs ^ sites/common_site_config.json | python -c "import sys, json; cfg = json.load(sys.stdin); print(cfg.get('db_host', ''))"` ]]; do
echo "Waiting for configurator to finish...";
sleep 5;
if (( `date +%s`-start > 120 )); then echo "Timed out"; break; fi
done;
echo "Starting site creation...";
bench new-site ${SITE_NAME:-erp.localhost} \
--no-mariadb-socket \
--db-root-password=${DB_ROOT_PASSWORD:-changeme123} \
--admin-password=${ADMIN_PASSWORD:-admin123} \
--install-app erpnext;
bench --site ${SITE_NAME:-erp.localhost} set-config allow_cors 1;
echo "Site created successfully.";
environment:
DB_ROOT_PASSWORD: ${DB_ROOT_PASSWORD:-changeme123}
ADMIN_PASSWORD: ${ADMIN_PASSWORD:-admin123}
SITE_NAME: ${SITE_NAME:-erp.localhost}
networks:
- frappe-net
depends_on:
configurator:
condition: service_completed_successfully
# ── Main backend ─────────────────────────────────────────────────────────────
backend:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
networks:
- frappe-net
depends_on:
create-site:
condition: service_completed_successfully
# ── WebSocket server ─────────────────────────────────────────────────────────
websocket:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
command: ["node", "/home/frappe/frappe-bench/apps/frappe/socketio.js"]
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
networks:
- frappe-net
depends_on:
create-site:
condition: service_completed_successfully
# ── Queue workers ─────────────────────────────────────────────────────────────
queue-short:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
command: ["bench", "worker", "--queue", "short,default"]
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
networks:
- frappe-net
depends_on:
create-site:
condition: service_completed_successfully
queue-long:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
command: ["bench", "worker", "--queue", "long,default,short"]
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
networks:
- frappe-net
depends_on:
create-site:
condition: service_completed_successfully
# ── Scheduler ─────────────────────────────────────────────────────────────────
scheduler:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
command: ["bench", "schedule"]
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
networks:
- frappe-net
depends_on:
create-site:
condition: service_completed_successfully
# ── Nginx (reverse proxy + static files) ─────────────────────────────────────
frontend:
image: docker.io/frappe/erpnext:${ERPNEXT_VERSION:-version-15}
restart: unless-stopped
command: ["nginx-entrypoint.sh"]
environment:
BACKEND: backend:8000
SOCKETIO: websocket:9000
FRAPPE_SITE_NAME_HEADER: ${SITE_NAME:-erp.localhost}
UPSTREAM_REAL_IP_ADDRESS: 127.0.0.1
UPSTREAM_REAL_IP_HEADER: X-Forwarded-For
UPSTREAM_REAL_IP_RECURSIVE: "off"
PROXY_READ_TIMEOUT: "120"
CLIENT_MAX_BODY_SIZE: "50m"
volumes:
- sites:/home/frappe/frappe-bench/sites
- logs:/home/frappe/frappe-bench/logs
ports:
- "${HTTP_PORT:-8080}:8080"
networks:
- frappe-net
depends_on:
backend:
condition: service_started
websocket:
condition: service_started
volumes:
db-data:
redis-cache-data:
redis-queue-data:
sites:
logs:
networks:
frappe-net:
driver: bridge

0
frappe_mcp/__init__.py Normal file
View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

55
frappe_mcp/audit_store.py Normal file
View File

@ -0,0 +1,55 @@
"""
Local audit store records every MCP tool call to a JSONL file on the MCP server.
This is MCP-level auditing (not Frappe-level). Complements Frappe's Version DocType.
"""
import json
import os
import time
from pathlib import Path
from datetime import datetime, timezone
_AUDIT_FILE = Path(os.environ.get("MCP_AUDIT_LOG", "frappe_mcp_audit.jsonl"))
def log_action(tool_name: str, arguments: dict, result_summary: str = "", risk: str = "low") -> str:
"""Append one audit record. Returns the record ID."""
record = {
"id": f"mcp-{int(time.time() * 1000)}",
"timestamp": datetime.now(timezone.utc).isoformat(),
"tool": tool_name,
"arguments": arguments,
"result_summary": result_summary[:200],
"risk": risk,
}
with _AUDIT_FILE.open("a", encoding="utf-8") as f:
f.write(json.dumps(record) + "\n")
return record["id"]
def read_audit_log(limit: int = 50, tool_filter: str = "") -> list[dict]:
if not _AUDIT_FILE.exists():
return []
records = []
with _AUDIT_FILE.open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
if tool_filter and tool_filter not in r.get("tool", ""):
continue
records.append(r)
except json.JSONDecodeError:
pass
return list(reversed(records))[:limit]
def clear_audit_log() -> int:
if not _AUDIT_FILE.exists():
return 0
with _AUDIT_FILE.open("r", encoding="utf-8") as f:
count = sum(1 for line in f if line.strip())
_AUDIT_FILE.unlink()
return count

View File

Binary file not shown.

Binary file not shown.

View File

@ -0,0 +1,144 @@
"""
Async HTTP client for Frappe REST API.
Handles API key/secret auth works with any remote Frappe instance including Docker.
"""
import json
import httpx
from typing import Any
from frappe_mcp.config import get_settings
class FrappeAPIError(Exception):
def __init__(self, message: str, status_code: int = 0, exc_type: str = ""):
super().__init__(message)
self.status_code = status_code
self.exc_type = exc_type
class FrappeClient:
def __init__(self):
self.settings = get_settings()
self.base_url = self.settings.frappe_url
self._auth_header = self._build_auth_header()
def _build_auth_header(self) -> dict[str, str]:
key = self.settings.frappe_api_key
secret = self.settings.frappe_api_secret
if not key or not secret:
raise FrappeAPIError("FRAPPE_API_KEY and FRAPPE_API_SECRET must be set in .env")
return {"Authorization": f"token {key}:{secret}"}
def _headers(self, extra: dict | None = None) -> dict[str, str]:
h = {**self._auth_header, "Content-Type": "application/json", "Accept": "application/json"}
if self.settings.frappe_site_name:
h["X-Frappe-Site-Name"] = self.settings.frappe_site_name
if extra:
h.update(extra)
return h
def _url(self, path: str) -> str:
return f"{self.base_url}{path}"
def _raise_for_frappe_error(self, data: dict) -> None:
if "exc_type" in data or ("message" in data and data.get("exc_type")):
raise FrappeAPIError(
data.get("message", "Unknown Frappe error"),
exc_type=data.get("exc_type", ""),
)
async def get(self, path: str, params: dict | None = None) -> Any:
async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
resp = await client.get(self._url(path), headers=self._headers(), params=params)
resp.raise_for_status()
data = resp.json()
self._raise_for_frappe_error(data)
return data.get("data", data)
def _handle_error_response(self, resp: httpx.Response) -> None:
"""Extract the real Frappe error message from 4xx/5xx responses."""
if resp.is_error:
try:
data = resp.json()
# Frappe puts the real error in _server_messages or exc
server_msgs = data.get("_server_messages", "")
exc = data.get("exc", "")
message = data.get("message", "")
if server_msgs:
import json as _json
try:
msgs = _json.loads(server_msgs)
parsed = [_json.loads(m).get("message", m) if isinstance(m, str) else m for m in msgs]
raise FrappeAPIError(
" | ".join(str(p) for p in parsed),
status_code=resp.status_code,
)
except (ValueError, TypeError):
pass
if exc:
last_line = [l for l in exc.strip().splitlines() if l.strip()]
raise FrappeAPIError(last_line[-1] if last_line else exc, status_code=resp.status_code)
if message:
raise FrappeAPIError(message, status_code=resp.status_code)
except FrappeAPIError:
raise
except Exception:
pass
resp.raise_for_status()
async def post(self, path: str, payload: dict | None = None) -> Any:
async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
resp = await client.post(self._url(path), headers=self._headers(), json=payload or {})
self._handle_error_response(resp)
data = resp.json()
self._raise_for_frappe_error(data)
return data.get("data", data)
async def put(self, path: str, payload: dict | None = None) -> Any:
async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
resp = await client.put(self._url(path), headers=self._headers(), json=payload or {})
self._handle_error_response(resp)
data = resp.json()
self._raise_for_frappe_error(data)
return data.get("data", data)
async def delete(self, path: str) -> Any:
async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
resp = await client.delete(self._url(path), headers=self._headers())
resp.raise_for_status()
data = resp.json()
self._raise_for_frappe_error(data)
return data.get("data", data)
# --- Frappe-specific convenience methods ---
async def call_method(self, method: str, **kwargs) -> Any:
"""Call a whitelisted Frappe server-side method."""
return await self.post(f"/api/method/{method}", payload=kwargs)
async def get_doc(self, doctype: str, name: str) -> dict:
return await self.get(f"/api/resource/{doctype}/{name}")
async def get_list(
self,
doctype: str,
fields: list[str] | None = None,
filters: list | None = None,
limit: int = 20,
order_by: str = "modified desc",
) -> list[dict]:
params: dict[str, Any] = {"limit": limit, "order_by": order_by}
if fields:
params["fields"] = json.dumps(fields)
if filters:
params["filters"] = json.dumps(filters)
return await self.get(f"/api/resource/{doctype}", params=params)
async def create_doc(self, doctype: str, data: dict) -> dict:
return await self.post(f"/api/resource/{doctype}", payload=data)
async def update_doc(self, doctype: str, name: str, data: dict) -> dict:
return await self.put(f"/api/resource/{doctype}/{name}", payload=data)
async def delete_doc(self, doctype: str, name: str) -> dict:
return await self.delete(f"/api/resource/{doctype}/{name}")

31
frappe_mcp/config.py Normal file
View File

@ -0,0 +1,31 @@
from pydantic_settings import BaseSettings
from pydantic import field_validator
from functools import lru_cache
class Settings(BaseSettings):
frappe_url: str = "http://localhost:8000"
frappe_api_key: str = ""
frappe_api_secret: str = ""
frappe_site_name: str = "" # optional: for multi-site Docker setups
# Safety: set to True to block destructive operations
read_only_mode: bool = False
# Timeout for REST calls in seconds
request_timeout: int = 30
@field_validator("frappe_url")
@classmethod
def strip_trailing_slash(cls, v: str) -> str:
return v.rstrip("/")
# Module activation — handled by module_registry, stored here so pydantic doesn't reject it
enabled_modules: str = ""
model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"}
@lru_cache
def get_settings() -> Settings:
return Settings()

82
frappe_mcp/healthcheck.py Normal file
View File

@ -0,0 +1,82 @@
"""
Health check run with: frappe-mcp --test-connection
Verifies connectivity, auth, and Frappe version before starting the MCP server.
"""
import asyncio
import sys
from frappe_mcp.client.frappe_api import FrappeClient, FrappeAPIError
from frappe_mcp.config import get_settings
async def run_health_check() -> bool:
settings = get_settings()
passed = True
print("\nFrappe MCP -- Connection Test")
print("=" * 45)
print(f" URL : {settings.frappe_url}")
print(f" API Key : {settings.frappe_api_key[:6]}..." if settings.frappe_api_key else " API Key : (not set)")
print(f" Site Name : {settings.frappe_site_name or '(default)'}")
print(f" Read-only : {settings.read_only_mode}")
print("=" * 45)
client = FrappeClient()
# 1. Ping / version check
print("\n[1/4] Checking Frappe version...", end=" ", flush=True)
try:
versions = await client.call_method("frappe.utils.change_log.get_versions")
frappe_ver = versions.get("frappe", {}).get("version", "unknown")
print(f"OK (Frappe {frappe_ver})")
except FrappeAPIError as e:
print(f"FAIL — {e}")
passed = False
except Exception as e:
print(f"FAIL — {e}")
passed = False
# 2. Auth check — fetch current user
print("[2/4] Verifying API credentials...", end=" ", flush=True)
try:
user = await client.call_method("frappe.auth.get_logged_user")
print(f"OK (logged in as: {user})")
except FrappeAPIError as e:
print(f"FAIL — {e}")
passed = False
except Exception as e:
print(f"FAIL — {e}")
passed = False
# 3. Read DocType list
print("[3/4] Testing DocType list access...", end=" ", flush=True)
try:
result = await client.get_list("DocType", fields=["name"], limit=1)
print(f"OK (found {len(result)} DocType(s))")
except Exception as e:
print(f"FAIL — {e}")
passed = False
# 4. Write check (skipped in read-only mode)
if settings.read_only_mode:
print("[4/4] Write check... SKIPPED (read_only_mode=true)")
else:
print("[4/4] Testing write permission (System Settings read)...", end=" ", flush=True)
try:
await client.get_doc("System Settings", "System Settings")
print("OK")
except Exception as e:
print(f"WARN — {e}")
print("=" * 45)
if passed:
print("PASS All checks passed. MCP server is ready.\n")
else:
print("FAIL Some checks failed. Fix the issues above before starting.\n")
return passed
def main_health_check():
ok = asyncio.run(run_health_check())
sys.exit(0 if ok else 1)

View File

@ -0,0 +1,137 @@
"""
Module Registry controls which tool modules are active.
Enable/disable via .env:
ENABLED_MODULES=documents,doctypes,users whitelist mode
MODULE_SCHEDULER=false disable one module
MODULE_GOVERNANCE=false disable another
Default: all modules enabled.
"""
import os
from typing import Callable
from mcp.types import Tool
# ── Import all tool modules ──────────────────────────────────────────────────
from frappe_mcp.tools import (
foundation,
doctypes,
documents,
document_inspect,
document_lifecycle,
custom_fields,
scripts,
workflow_tools,
users,
admin,
analytics,
print_formats,
email_templates,
property_setters,
naming_series,
bulk_ops,
files,
activity,
dashboards,
webhooks,
reports,
scheduler,
translations,
business_actions,
governance,
)
# ── Module definitions ───────────────────────────────────────────────────────
# (module_key, module_object, description)
ALL_MODULES: list[tuple[str, object, str]] = [
# Level 1 — Foundation
("foundation", foundation, "L1: ping, session_info, doctype_meta, permissions, modules"),
# Level 2 — Read
("doctypes", doctypes, "L1-2: DocType CRUD — create, read, update, list"),
("documents", documents, "L2-3: Document CRUD — create, read, update, delete, submit, cancel"),
("document_inspect", document_inspect, "L2: Search, children, linked docs, timeline, count, attachments"),
# Level 3-4 — Write + Lifecycle
("document_lifecycle", document_lifecycle, "L3-4: Child rows, rename, amend, duplicate, status, draft save"),
("custom_fields", custom_fields, "L2/9: Custom Fields — add, edit, remove, list"),
("scripts", scripts, "L9: Server Scripts + Client Scripts"),
# Level 5 — Workflow
("workflow_tools", workflow_tools, "L5: Workflows, transitions, approvals"),
# Level 6 — Reporting
("analytics", analytics, "L6: Aggregate, dashboard data, PDF render, number cards"),
("reports", reports, "L6/9: Query/Script Report CRUD"),
("print_formats", print_formats, "L6: Print Format templates"),
# Level 7 — Bulk
("bulk_ops", bulk_ops, "L7: Bulk create/update/delete/submit/cancel/assign/tag"),
# Level 8 — Business Actions
("business_actions", business_actions, "L8: ERPNext shortcuts — Sales, Buying, Stock, HR, Support, Projects"),
# Level 9 — Admin / System Intelligence
("users", users, "L9: Users, Roles, DocType permissions"),
("admin", admin, "L9: Cache, system settings, SQL, workflows, scheduler"),
("property_setters", property_setters, "L9: Property Setters + Customize Form"),
("naming_series", naming_series, "L9: Naming Series — patterns and counters"),
("files", files, "L2/9: File Manager — upload, list, delete, move"),
("activity", activity, "L2/9: Comments, tags, assignments, ToDo, error logs"),
("dashboards", dashboards, "L6/9: Dashboards, charts, number cards, workspaces"),
("webhooks", webhooks, "L9: Webhooks + API Keys"),
("email_templates", email_templates, "L9: Email Templates + Notifications"),
("translations", translations, "L9: Translations, Assignment Rules, User Permissions"),
("scheduler", scheduler, "L9: Scheduled Jobs + Background Jobs"),
# Level 10 — Safety & Governance
("governance", governance, "L10: Dry run, validate, risk score, audit log, rollback"),
]
def _is_module_enabled(key: str) -> bool:
"""
Resolution order (first match wins):
1. MODULE_<KEY>=false disabled
2. MODULE_<KEY>=true enabled
3. ENABLED_MODULES=a,b whitelist only listed enabled
4. Default: enabled
"""
env_key = f"MODULE_{key.upper()}"
explicit = os.environ.get(env_key, "").strip().lower()
if explicit == "false":
return False
if explicit == "true":
return True
enabled_list = os.environ.get("ENABLED_MODULES", "").strip()
if enabled_list:
return key in [m.strip() for m in enabled_list.split(",")]
return True
def get_enabled_tools() -> list[Tool]:
tools: list[Tool] = []
for key, module, _ in ALL_MODULES:
if _is_module_enabled(key):
tools.extend(module.tools())
return tools
def get_enabled_handlers() -> dict[str, Callable]:
handlers: dict[str, Callable] = {}
for key, module, _ in ALL_MODULES:
if _is_module_enabled(key):
handlers.update(module.handlers())
return handlers
def print_module_status():
print("\nFrappe MCP — Module Status")
print("=" * 70)
total_on = 0
total_all = 0
for key, module, desc in ALL_MODULES:
enabled = _is_module_enabled(key)
count = len(module.tools())
total_all += count
if enabled:
total_on += count
status = "ON " if enabled else "OFF"
print(f" [{status}] {key:<22} ({count:>2} tools) {desc}")
print("=" * 70)
print(f" Active: {total_on} tools | Total available: {total_all} tools\n")

103
frappe_mcp/server.py Normal file
View File

@ -0,0 +1,103 @@
"""
Frappe MCP Server entry point.
Modules can be individually enabled/disabled via ENABLED_MODULES in .env.
Transports:
(default) stdio for local Claude Desktop use
--sse HTTP/SSE for VPS hosting (requires starlette + uvicorn)
"""
import asyncio
import os
import sys
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from frappe_mcp.module_registry import get_enabled_tools, get_enabled_handlers
app = Server("frappe-mcp")
@app.list_tools()
async def list_tools() -> list[Tool]:
return get_enabled_tools()
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
handlers = get_enabled_handlers()
handler = handlers.get(name)
if not handler:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
try:
result = await handler(arguments)
return [TextContent(type="text", text=result)]
except Exception as e:
return [TextContent(type="text", text=f"Error: {e}")]
def main():
if "--test-connection" in sys.argv:
from frappe_mcp.healthcheck import run_health_check
ok = asyncio.run(run_health_check())
sys.exit(0 if ok else 1)
if "--list-modules" in sys.argv:
from frappe_mcp.module_registry import print_module_status
print_module_status()
sys.exit(0)
if "--sse" in sys.argv:
_run_sse()
return
asyncio.run(_run_stdio())
async def _run_stdio():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
def _run_sse():
try:
import uvicorn
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Mount, Route
from mcp.server.sse import SseServerTransport
except ImportError:
print("SSE dependencies missing. Run: pip install 'frappe-mcp[sse]'", file=sys.stderr)
sys.exit(1)
host = os.environ.get("MCP_HOST", "0.0.0.0")
port = int(os.environ.get("MCP_PORT", "8001"))
bearer_token = os.environ.get("MCP_BEARER_TOKEN", "")
sse = SseServerTransport("/messages/")
async def handle_sse(request: Request) -> Response:
if bearer_token:
auth = request.headers.get("Authorization", "")
if auth != f"Bearer {bearer_token}":
return Response("Unauthorized", status_code=401)
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
return Response()
starlette_app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
]
)
print(f"Frappe MCP SSE server starting on {host}:{port}")
uvicorn.run(starlette_app, host=host, port=port)
if __name__ == "__main__":
main()

View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@ -0,0 +1,390 @@
"""
Activity, collaboration, and system log tools.
Covers: Comments, Tags, Assignments, ToDo, Activity Log, Error Log, Document Follow.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
# --- Comments ---
Tool(
name="frappe_add_comment",
description="Add a comment to any document.",
inputSchema={
"type": "object",
"required": ["reference_doctype", "reference_name", "content"],
"properties": {
"reference_doctype": {"type": "string"},
"reference_name": {"type": "string"},
"content": {"type": "string"},
"comment_type": {
"type": "string",
"enum": ["Comment", "Info", "Warning", "Error", "Workflow", "Label"],
"default": "Comment",
},
},
},
),
Tool(
name="frappe_get_comments",
description="Get all comments for a document.",
inputSchema={
"type": "object",
"required": ["reference_doctype", "reference_name"],
"properties": {
"reference_doctype": {"type": "string"},
"reference_name": {"type": "string"},
},
},
),
# --- Tags ---
Tool(
name="frappe_add_tag",
description="Add a tag to a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "tag"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"tag": {"type": "string"},
},
},
),
Tool(
name="frappe_remove_tag",
description="Remove a tag from a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "tag"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"tag": {"type": "string"},
},
},
),
Tool(
name="frappe_get_tags",
description="Get all tags on a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
# --- Assignments ---
Tool(
name="frappe_assign_document",
description="Assign a document to one or more users.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "assign_to"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"assign_to": {
"type": "array",
"items": {"type": "string"},
"description": "List of user emails to assign",
},
"description": {"type": "string"},
"due_date": {"type": "string", "description": "YYYY-MM-DD"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High"], "default": "Medium"},
"notify": {"type": "integer", "default": 0},
},
},
),
Tool(
name="frappe_remove_assignment",
description="Remove assignment of a document from a user.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "assign_to"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"assign_to": {"type": "string", "description": "User email to unassign"},
},
},
),
Tool(
name="frappe_get_assignments",
description="Get all active assignments for a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
# --- ToDo ---
Tool(
name="frappe_create_todo",
description="Create a ToDo item (task) optionally linked to a document.",
inputSchema={
"type": "object",
"required": ["description"],
"properties": {
"description": {"type": "string"},
"assigned_by": {"type": "string"},
"owner": {"type": "string", "description": "Assigned to user"},
"reference_type": {"type": "string"},
"reference_name": {"type": "string"},
"due_date": {"type": "string"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High"], "default": "Medium"},
"status": {"type": "string", "enum": ["Open", "Closed"], "default": "Open"},
},
},
),
# --- Activity Log ---
Tool(
name="frappe_get_activity_log",
description="Get the activity log for a specific document (who changed what and when).",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"limit": {"type": "integer", "default": 20},
},
},
),
# --- Error Logs ---
Tool(
name="frappe_get_error_logs",
description="Get recent system error logs.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "default": 20},
"method": {"type": "string", "description": "Filter by method name"},
},
},
),
Tool(
name="frappe_clear_error_logs",
description="Clear all error logs.",
inputSchema={"type": "object", "properties": {}},
),
# --- Document Follow ---
Tool(
name="frappe_follow_document",
description="Follow a document to receive notifications on changes.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_unfollow_document",
description="Unfollow a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_add_comment": _add_comment,
"frappe_get_comments": _get_comments,
"frappe_add_tag": _add_tag,
"frappe_remove_tag": _remove_tag,
"frappe_get_tags": _get_tags,
"frappe_assign_document": _assign_document,
"frappe_remove_assignment": _remove_assignment,
"frappe_get_assignments": _get_assignments,
"frappe_create_todo": _create_todo,
"frappe_get_activity_log": _get_activity_log,
"frappe_get_error_logs": _get_error_logs,
"frappe_clear_error_logs": _clear_error_logs,
"frappe_follow_document": _follow_document,
"frappe_unfollow_document": _unfollow_document,
}
async def _add_comment(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.client.add_comment",
reference_doctype=args["reference_doctype"],
reference_name=args["reference_name"],
content=args["content"],
comment_email="",
comment_by="",
)
return json.dumps(result, indent=2)
async def _get_comments(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Comment",
fields=["name", "comment_type", "comment_by", "content", "creation"],
filters=[
["reference_doctype", "=", args["reference_doctype"]],
["reference_name", "=", args["reference_name"]],
],
limit=50,
order_by="creation asc",
)
return json.dumps(result, indent=2)
async def _add_tag(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.doctype.tag.tag.add_tag",
tag=args["tag"],
dt=args["doctype"],
dn=args["name"],
)
return json.dumps(result, indent=2)
async def _remove_tag(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.doctype.tag.tag.remove_tag",
tag=args["tag"],
dt=args["doctype"],
dn=args["name"],
)
return json.dumps(result, indent=2)
async def _get_tags(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.doctype.tag.tag.get_tags",
dt=args["doctype"],
dn=args["name"],
)
return json.dumps(result, indent=2)
async def _assign_document(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.form.assign_to.add",
args={
"doctype": args["doctype"],
"name": args["name"],
"assign_to": args["assign_to"],
"description": args.get("description", ""),
"due_date": args.get("due_date", ""),
"priority": args.get("priority", "Medium"),
"notify": args.get("notify", 0),
},
)
return json.dumps(result, indent=2)
async def _remove_assignment(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.form.assign_to.remove",
doctype=args["doctype"],
name=args["name"],
assign_to=args["assign_to"],
)
return json.dumps(result, indent=2)
async def _get_assignments(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"ToDo",
fields=["name", "owner", "description", "due_date", "priority", "status"],
filters=[
["reference_type", "=", args["doctype"]],
["reference_name", "=", args["name"]],
["status", "=", "Open"],
],
limit=20,
)
return json.dumps(result, indent=2)
async def _create_todo(args: dict) -> str:
client = FrappeClient()
payload = {k: v for k, v in args.items()}
result = await client.create_doc("ToDo", payload)
return json.dumps(result, indent=2)
async def _get_activity_log(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Version",
fields=["name", "owner", "creation", "data"],
filters=[
["ref_doctype", "=", args["doctype"]],
["docname", "=", args["name"]],
],
limit=args.get("limit", 20),
order_by="creation desc",
)
return json.dumps(result, indent=2)
async def _get_error_logs(args: dict) -> str:
client = FrappeClient()
filters = []
if method := args.get("method"):
filters.append(["method", "like", f"%{method}%"])
result = await client.get_list(
"Error Log",
fields=["name", "method", "error", "creation"],
filters=filters if filters else None,
limit=args.get("limit", 20),
order_by="creation desc",
)
return json.dumps(result, indent=2)
async def _clear_error_logs(args: dict) -> str:
client = FrappeClient()
result = await client.call_method("frappe.core.doctype.error_log.error_log.clear_error_logs")
return json.dumps(result, indent=2)
async def _follow_document(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.form.document_follow.follow_document",
doctype=args["doctype"],
doc_name=args["name"],
)
return json.dumps(result, indent=2)
async def _unfollow_document(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.form.document_follow.unfollow_document",
doctype=args["doctype"],
doc_name=args["name"],
)
return json.dumps(result, indent=2)

185
frappe_mcp/tools/admin.py Normal file
View File

@ -0,0 +1,185 @@
"""Admin / bench tools — cache, migrate, app management, system settings."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
from frappe_mcp.config import get_settings
def tools() -> list[Tool]:
return [
Tool(
name="frappe_clear_cache",
description="Clear Frappe's server-side cache.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_get_installed_apps",
description="List all Frappe apps installed on the site.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_get_system_settings",
description="Get the current System Settings document.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_update_system_settings",
description="Update System Settings fields.",
inputSchema={
"type": "object",
"required": ["updates"],
"properties": {
"updates": {
"type": "object",
"description": "Fields to update e.g. {\"language\": \"en\", \"time_zone\": \"Asia/Kolkata\"}",
},
},
},
),
Tool(
name="frappe_run_report",
description="Run a Script Report or Query Report and return results.",
inputSchema={
"type": "object",
"required": ["report_name"],
"properties": {
"report_name": {"type": "string"},
"filters": {
"type": "object",
"description": "Report filters as key-value pairs",
},
},
},
),
Tool(
name="frappe_execute_query",
description=(
"Execute a read-only SQL query on the Frappe database via frappe.db.sql. "
"Only SELECT statements are allowed."
),
inputSchema={
"type": "object",
"required": ["query"],
"properties": {
"query": {"type": "string", "description": "SQL SELECT query"},
"values": {
"type": "array",
"description": "Parameterized values for %s placeholders",
},
},
},
),
Tool(
name="frappe_get_site_info",
description="Get basic site information: Frappe version, installed apps, site config.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_create_workflow",
description="Create a Workflow for a DocType with states and transitions.",
inputSchema={
"type": "object",
"required": ["workflow_name", "document_type", "states", "transitions"],
"properties": {
"workflow_name": {"type": "string"},
"document_type": {"type": "string"},
"is_active": {"type": "integer", "default": 1},
"states": {
"type": "array",
"description": "List of workflow state objects with: state, doc_status, allow_edit",
"items": {"type": "object"},
},
"transitions": {
"type": "array",
"description": "List of transition objects with: state, action, next_state, allowed",
"items": {"type": "object"},
},
"send_email_alert": {"type": "integer", "default": 0},
},
},
),
]
def handlers() -> dict:
return {
"frappe_clear_cache": _clear_cache,
"frappe_get_installed_apps": _get_installed_apps,
"frappe_get_system_settings": _get_system_settings,
"frappe_update_system_settings": _update_system_settings,
"frappe_run_report": _run_report,
"frappe_execute_query": _execute_query,
"frappe_get_site_info": _get_site_info,
"frappe_create_workflow": _create_workflow,
}
async def _clear_cache(args: dict) -> str:
client = FrappeClient()
result = await client.call_method("frappe.sessions.clear")
return json.dumps(result, indent=2)
async def _get_installed_apps(args: dict) -> str:
client = FrappeClient()
result = await client.call_method("frappe.utils.change_log.get_versions")
return json.dumps(result, indent=2)
async def _get_system_settings(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("System Settings", "System Settings")
return json.dumps(result, indent=2)
async def _update_system_settings(args: dict) -> str:
if get_settings().read_only_mode:
return "Error: read_only_mode is enabled."
client = FrappeClient()
result = await client.update_doc("System Settings", "System Settings", args["updates"])
return json.dumps(result, indent=2)
async def _run_report(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.query_report.run",
report_name=args["report_name"],
filters=args.get("filters", {}),
)
return json.dumps(result, indent=2)
async def _execute_query(args: dict) -> str:
query = args["query"].strip().lower()
if not query.startswith("select"):
return "Error: only SELECT queries are allowed."
client = FrappeClient()
result = await client.call_method(
"frappe.client.get_list",
doctype="__query__",
query=args["query"],
values=args.get("values", []),
)
return json.dumps(result, indent=2)
async def _get_site_info(args: dict) -> str:
client = FrappeClient()
result = await client.call_method("frappe.utils.change_log.get_versions")
return json.dumps(result, indent=2)
async def _create_workflow(args: dict) -> str:
client = FrappeClient()
payload = {
"workflow_name": args["workflow_name"],
"document_type": args["document_type"],
"is_active": args.get("is_active", 1),
"send_email_alert": args.get("send_email_alert", 0),
"states": args["states"],
"transitions": args["transitions"],
}
result = await client.create_doc("Workflow", payload)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,176 @@
"""
Level 6 Reporting and analytics tools.
Aggregate queries, dashboard data, PDF rendering.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_aggregate_documents",
description=(
"Aggregate document data: sum, avg, min, max, or count grouped by a field. "
"e.g. total sales per customer, count of issues per status."
),
inputSchema={
"type": "object",
"required": ["doctype", "function", "field"],
"properties": {
"doctype": {"type": "string"},
"function": {
"type": "string",
"enum": ["sum", "avg", "min", "max", "count"],
},
"field": {"type": "string", "description": "Field to aggregate e.g. 'grand_total'"},
"group_by": {"type": "string", "description": "Field to group by e.g. 'customer'"},
"filters": {"type": "array"},
"limit": {"type": "integer", "default": 50},
},
},
),
Tool(
name="frappe_get_dashboard_data",
description=(
"Pull data for a Dashboard Chart — returns the chart's data points "
"for display as a summary or KPI."
),
inputSchema={
"type": "object",
"required": ["chart_name"],
"properties": {
"chart_name": {"type": "string"},
"filters": {"type": "object"},
"refresh": {"type": "boolean", "default": False},
},
},
),
Tool(
name="frappe_render_document_pdf",
description=(
"Generate a PDF for a document using a Print Format. "
"Returns the PDF download URL on the Frappe server."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"print_format": {"type": "string", "description": "Print Format name (optional, uses default if empty)"},
"letterhead": {"type": "string"},
"language": {"type": "string", "default": "en"},
},
},
),
Tool(
name="frappe_get_number_card_value",
description="Get the current value of a Number Card (single KPI metric).",
inputSchema={
"type": "object",
"required": ["card_name"],
"properties": {
"card_name": {"type": "string"},
"filters": {"type": "object"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_aggregate_documents": _aggregate_documents,
"frappe_get_dashboard_data": _get_dashboard_data,
"frappe_render_document_pdf": _render_document_pdf,
"frappe_get_number_card_value": _get_number_card_value,
}
async def _aggregate_documents(args: dict) -> str:
client = FrappeClient()
func = args["function"].upper()
field = args["field"]
group_by = args.get("group_by")
filters = args.get("filters")
if group_by:
sql = f"SELECT `{group_by}`, {func}(`{field}`) as value FROM `tab{args['doctype']}`"
conditions = []
if filters and isinstance(filters, list):
for f in filters:
if len(f) >= 3:
conditions.append(f"`{f[0]}` {f[1]} '{f[2]}'")
if conditions:
sql += " WHERE " + " AND ".join(conditions)
sql += f" GROUP BY `{group_by}` ORDER BY value DESC LIMIT {args.get('limit', 50)}"
result = await client.call_method("frappe.client.get_list", doctype="__query__", query=sql)
else:
sql = f"SELECT {func}(`{field}`) as value FROM `tab{args['doctype']}`"
conditions = []
if filters and isinstance(filters, list):
for f in filters:
if len(f) >= 3:
conditions.append(f"`{f[0]}` {f[1]} '{f[2]}'")
if conditions:
sql += " WHERE " + " AND ".join(conditions)
result = await client.call_method(
"frappe.desk.query_report.run",
report_name="__inline__",
filters={},
)
# Fallback: use get_list with aggregation via API
result = await client.call_method(
"frappe.client.get_value",
doctype=args["doctype"],
filters=filters,
fieldname=f"{func.lower()}({field})",
)
return json.dumps({"function": func, "field": field, "group_by": group_by, "result": result}, indent=2)
async def _get_dashboard_data(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.doctype.dashboard_chart.dashboard_chart.get",
chart_name=args["chart_name"],
filters=json.dumps(args.get("filters", {})),
refresh=args.get("refresh", False),
)
return json.dumps(result, indent=2)
async def _render_document_pdf(args: dict) -> str:
client = FrappeClient()
params = {
"doctype": args["doctype"],
"name": args["name"],
"print_format": args.get("print_format", ""),
"letterhead": args.get("letterhead", ""),
"lang": args.get("language", "en"),
}
pdf_url = (
f"{client.base_url}/api/method/frappe.utils.pdf.download_pdf"
f"?doctype={params['doctype']}&name={params['name']}"
f"&format={params['print_format']}&lang={params['lang']}"
)
return json.dumps({
"pdf_url": pdf_url,
"doctype": args["doctype"],
"name": args["name"],
"print_format": args.get("print_format", "default"),
"note": "Open pdf_url in browser or download directly from Frappe server",
}, indent=2)
async def _get_number_card_value(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.desk.doctype.number_card.number_card.get_result",
doc={"name": args["card_name"]},
filters=json.dumps(args.get("filters", {})),
)
return json.dumps({"card": args["card_name"], "value": result}, indent=2)

View File

@ -0,0 +1,337 @@
"""Bulk operations — bulk create, update, delete documents and data import/export."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
from frappe_mcp.config import get_settings
def tools() -> list[Tool]:
return [
Tool(
name="frappe_bulk_create_documents",
description="Create multiple documents of the same DocType in one call.",
inputSchema={
"type": "object",
"required": ["doctype", "documents"],
"properties": {
"doctype": {"type": "string"},
"documents": {
"type": "array",
"items": {"type": "object"},
"description": "List of document field dicts to create",
},
},
},
),
Tool(
name="frappe_bulk_update_documents",
description="Update the same field(s) on multiple documents at once.",
inputSchema={
"type": "object",
"required": ["doctype", "names", "updates"],
"properties": {
"doctype": {"type": "string"},
"names": {
"type": "array",
"items": {"type": "string"},
"description": "List of document names to update",
},
"updates": {
"type": "object",
"description": "Fields and values to apply to all documents",
},
},
},
),
Tool(
name="frappe_bulk_delete_documents",
description="Delete multiple documents. Blocked in read-only mode.",
inputSchema={
"type": "object",
"required": ["doctype", "names"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
},
},
),
Tool(
name="frappe_bulk_submit_documents",
description="Submit multiple documents in one call.",
inputSchema={
"type": "object",
"required": ["doctype", "names"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
},
},
),
Tool(
name="frappe_import_data",
description=(
"Import documents via JSON array. Equivalent to Frappe's Data Import tool. "
"Use 'Insert' to create new records or 'Update' to update existing ones."
),
inputSchema={
"type": "object",
"required": ["doctype", "data"],
"properties": {
"doctype": {"type": "string"},
"data": {
"type": "array",
"items": {"type": "object"},
"description": "Array of document objects to import",
},
"import_type": {
"type": "string",
"enum": ["Insert New Records", "Update Existing Records"],
"default": "Insert New Records",
},
},
},
),
Tool(
name="frappe_export_data",
description="Export documents from a DocType as a JSON list.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"fields": {
"type": "array",
"items": {"type": "string"},
"description": "Fields to export. Empty = all fields.",
},
"filters": {"type": "array"},
"limit": {"type": "integer", "default": 500},
},
},
),
Tool(
name="frappe_bulk_cancel_documents",
description="Cancel multiple submitted documents.",
inputSchema={
"type": "object",
"required": ["doctype", "names"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
},
},
),
Tool(
name="frappe_bulk_assign_documents",
description="Assign multiple documents to a user.",
inputSchema={
"type": "object",
"required": ["doctype", "names", "assign_to"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
"assign_to": {"type": "array", "items": {"type": "string"}, "description": "List of user emails"},
"description": {"type": "string"},
"due_date": {"type": "string"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High"], "default": "Medium"},
},
},
),
Tool(
name="frappe_bulk_add_tags",
description="Add the same tag to multiple documents.",
inputSchema={
"type": "object",
"required": ["doctype", "names", "tag"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
"tag": {"type": "string"},
},
},
),
Tool(
name="frappe_bulk_add_comment",
description="Add the same comment to multiple documents.",
inputSchema={
"type": "object",
"required": ["doctype", "names", "content"],
"properties": {
"doctype": {"type": "string"},
"names": {"type": "array", "items": {"type": "string"}},
"content": {"type": "string"},
"comment_type": {"type": "string", "default": "Comment"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_bulk_create_documents": _bulk_create,
"frappe_bulk_update_documents": _bulk_update,
"frappe_bulk_delete_documents": _bulk_delete,
"frappe_bulk_submit_documents": _bulk_submit,
"frappe_import_data": _import_data,
"frappe_export_data": _export_data,
"frappe_bulk_cancel_documents": _bulk_cancel,
"frappe_bulk_assign_documents": _bulk_assign,
"frappe_bulk_add_tags": _bulk_add_tags,
"frappe_bulk_add_comment": _bulk_add_comment,
}
async def _bulk_create(args: dict) -> str:
client = FrappeClient()
results = []
errors = []
for doc in args["documents"]:
try:
data = {"doctype": args["doctype"], **doc}
result = await client.create_doc(args["doctype"], data)
results.append(result.get("name") if isinstance(result, dict) else result)
except Exception as e:
errors.append({"doc": doc, "error": str(e)})
return json.dumps({"created": results, "errors": errors}, indent=2)
async def _bulk_update(args: dict) -> str:
client = FrappeClient()
results = []
errors = []
for name in args["names"]:
try:
result = await client.update_doc(args["doctype"], name, args["updates"])
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"updated": results, "errors": errors}, indent=2)
async def _bulk_delete(args: dict) -> str:
if get_settings().read_only_mode:
return "Error: read_only_mode is enabled. Deletion blocked."
client = FrappeClient()
results = []
errors = []
for name in args["names"]:
try:
await client.delete_doc(args["doctype"], name)
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"deleted": results, "errors": errors}, indent=2)
async def _bulk_submit(args: dict) -> str:
client = FrappeClient()
results = []
errors = []
for name in args["names"]:
try:
await client.update_doc(args["doctype"], name, {"docstatus": 1})
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"submitted": results, "errors": errors}, indent=2)
async def _import_data(args: dict) -> str:
client = FrappeClient()
results = []
errors = []
is_update = args.get("import_type", "Insert New Records") == "Update Existing Records"
for doc in args["data"]:
try:
name = doc.get("name")
if is_update and name:
result = await client.update_doc(args["doctype"], name, doc)
else:
data = {"doctype": args["doctype"], **doc}
result = await client.create_doc(args["doctype"], data)
results.append(result.get("name") if isinstance(result, dict) else str(result))
except Exception as e:
errors.append({"doc": doc.get("name", str(doc)[:40]), "error": str(e)})
return json.dumps({"imported": len(results), "names": results, "errors": errors}, indent=2)
async def _export_data(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
args["doctype"],
fields=args.get("fields") or ["*"],
filters=args.get("filters"),
limit=args.get("limit", 500),
)
return json.dumps(result, indent=2)
async def _bulk_cancel(args: dict) -> str:
client = FrappeClient()
results, errors = [], []
for name in args["names"]:
try:
await client.update_doc(args["doctype"], name, {"docstatus": 2})
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"cancelled": results, "errors": errors}, indent=2)
async def _bulk_assign(args: dict) -> str:
client = FrappeClient()
results, errors = [], []
for name in args["names"]:
try:
await client.call_method(
"frappe.desk.form.assign_to.add",
args={
"doctype": args["doctype"],
"name": name,
"assign_to": args["assign_to"],
"description": args.get("description", ""),
"due_date": args.get("due_date", ""),
"priority": args.get("priority", "Medium"),
"notify": 0,
},
)
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"assigned": results, "errors": errors}, indent=2)
async def _bulk_add_tags(args: dict) -> str:
client = FrappeClient()
results, errors = [], []
for name in args["names"]:
try:
await client.call_method(
"frappe.desk.doctype.tag.tag.add_tag",
tag=args["tag"], dt=args["doctype"], dn=name,
)
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"tagged": results, "errors": errors}, indent=2)
async def _bulk_add_comment(args: dict) -> str:
client = FrappeClient()
results, errors = [], []
for name in args["names"]:
try:
await client.call_method(
"frappe.client.add_comment",
reference_doctype=args["doctype"],
reference_name=name,
content=args["content"],
comment_email="",
comment_by="",
)
results.append(name)
except Exception as e:
errors.append({"name": name, "error": str(e)})
return json.dumps({"commented": results, "errors": errors}, indent=2)

View File

@ -0,0 +1,477 @@
"""
Level 8 Business action tools (ERPNext shortcuts).
High-value operations: convert documents, approve, transfer stock, etc.
These call ERPNext's built-in mapper/make functions.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
# ── Sales / CRM ──────────────────────────────────────────────────────
Tool(
name="erpnext_create_quotation_from_opportunity",
description="Create a Quotation from an Opportunity.",
inputSchema={
"type": "object",
"required": ["opportunity_name"],
"properties": {"opportunity_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_sales_order_from_quotation",
description="Create a Sales Order from a Quotation.",
inputSchema={
"type": "object",
"required": ["quotation_name"],
"properties": {"quotation_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_sales_invoice_from_sales_order",
description="Create a Sales Invoice from a Sales Order.",
inputSchema={
"type": "object",
"required": ["sales_order_name"],
"properties": {"sales_order_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_delivery_note_from_sales_order",
description="Create a Delivery Note from a Sales Order.",
inputSchema={
"type": "object",
"required": ["sales_order_name"],
"properties": {"sales_order_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_payment_entry_for_invoice",
description="Create a Payment Entry for a Sales Invoice or Purchase Invoice.",
inputSchema={
"type": "object",
"required": ["invoice_doctype", "invoice_name"],
"properties": {
"invoice_doctype": {"type": "string", "enum": ["Sales Invoice", "Purchase Invoice"]},
"invoice_name": {"type": "string"},
},
},
),
# ── Buying ───────────────────────────────────────────────────────────
Tool(
name="erpnext_create_purchase_order_from_supplier_quotation",
description="Create a Purchase Order from a Supplier Quotation.",
inputSchema={
"type": "object",
"required": ["supplier_quotation_name"],
"properties": {"supplier_quotation_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_purchase_receipt_from_purchase_order",
description="Create a Purchase Receipt from a Purchase Order.",
inputSchema={
"type": "object",
"required": ["purchase_order_name"],
"properties": {"purchase_order_name": {"type": "string"}},
},
),
Tool(
name="erpnext_create_purchase_invoice_from_purchase_receipt",
description="Create a Purchase Invoice from a Purchase Receipt.",
inputSchema={
"type": "object",
"required": ["purchase_receipt_name"],
"properties": {"purchase_receipt_name": {"type": "string"}},
},
),
# ── Stock ────────────────────────────────────────────────────────────
Tool(
name="erpnext_get_item_stock_balance",
description="Get current stock balance for an item, optionally at a specific warehouse.",
inputSchema={
"type": "object",
"required": ["item_code"],
"properties": {
"item_code": {"type": "string"},
"warehouse": {"type": "string"},
"posting_date": {"type": "string", "description": "YYYY-MM-DD (defaults to today)"},
},
},
),
Tool(
name="erpnext_transfer_stock",
description="Transfer stock between warehouses via a Material Transfer Stock Entry.",
inputSchema={
"type": "object",
"required": ["item_code", "qty", "from_warehouse", "to_warehouse"],
"properties": {
"item_code": {"type": "string"},
"qty": {"type": "number"},
"from_warehouse": {"type": "string"},
"to_warehouse": {"type": "string"},
"posting_date": {"type": "string"},
"remarks": {"type": "string"},
},
},
),
Tool(
name="erpnext_create_stock_entry",
description=(
"Create a Stock Entry of any type: "
"Material Issue, Material Receipt, Material Transfer, Manufacture, Repack."
),
inputSchema={
"type": "object",
"required": ["stock_entry_type", "items"],
"properties": {
"stock_entry_type": {
"type": "string",
"enum": ["Material Issue", "Material Receipt", "Material Transfer",
"Material Transfer for Manufacture", "Manufacture", "Repack", "Send to Subcontractor"],
},
"items": {
"type": "array",
"description": "List of {item_code, qty, s_warehouse, t_warehouse} objects",
"items": {"type": "object"},
},
"posting_date": {"type": "string"},
"remarks": {"type": "string"},
},
},
),
# ── HR ───────────────────────────────────────────────────────────────
Tool(
name="erpnext_approve_leave_application",
description="Approve a Leave Application.",
inputSchema={
"type": "object",
"required": ["leave_application_name"],
"properties": {"leave_application_name": {"type": "string"}},
},
),
Tool(
name="erpnext_reject_leave_application",
description="Reject a Leave Application.",
inputSchema={
"type": "object",
"required": ["leave_application_name"],
"properties": {
"leave_application_name": {"type": "string"},
"reason": {"type": "string"},
},
},
),
Tool(
name="erpnext_mark_attendance",
description="Mark attendance for an employee on a specific date.",
inputSchema={
"type": "object",
"required": ["employee", "attendance_date", "status"],
"properties": {
"employee": {"type": "string"},
"attendance_date": {"type": "string", "description": "YYYY-MM-DD"},
"status": {"type": "string", "enum": ["Present", "Absent", "Half Day", "Work From Home", "On Leave"]},
"shift": {"type": "string"},
},
},
),
Tool(
name="erpnext_approve_expense_claim",
description="Approve an Expense Claim.",
inputSchema={
"type": "object",
"required": ["expense_claim_name"],
"properties": {"expense_claim_name": {"type": "string"}},
},
),
# ── Support ──────────────────────────────────────────────────────────
Tool(
name="erpnext_create_issue",
description="Create a new Support Issue.",
inputSchema={
"type": "object",
"required": ["subject"],
"properties": {
"subject": {"type": "string"},
"customer": {"type": "string"},
"description": {"type": "string"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High", "Urgent"], "default": "Medium"},
"issue_type": {"type": "string"},
"raised_by": {"type": "string"},
},
},
),
Tool(
name="erpnext_close_issue",
description="Close a Support Issue.",
inputSchema={
"type": "object",
"required": ["issue_name"],
"properties": {
"issue_name": {"type": "string"},
"resolution_details": {"type": "string"},
},
},
),
# ── Projects ─────────────────────────────────────────────────────────
Tool(
name="erpnext_create_task",
description="Create a Task in a Project.",
inputSchema={
"type": "object",
"required": ["subject"],
"properties": {
"subject": {"type": "string"},
"project": {"type": "string"},
"assigned_to": {"type": "string"},
"exp_start_date": {"type": "string"},
"exp_end_date": {"type": "string"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High", "Urgent"]},
"description": {"type": "string"},
},
},
),
Tool(
name="erpnext_update_task_status",
description="Update the status of a Task.",
inputSchema={
"type": "object",
"required": ["task_name", "status"],
"properties": {
"task_name": {"type": "string"},
"status": {"type": "string", "enum": ["Open", "Working", "Pending Review", "Overdue", "Template", "Cancelled", "Completed"]},
},
},
),
Tool(
name="erpnext_log_timesheet_entry",
description="Log a timesheet entry for an employee on a task or project.",
inputSchema={
"type": "object",
"required": ["employee", "from_time", "to_time"],
"properties": {
"employee": {"type": "string"},
"from_time": {"type": "string", "description": "ISO datetime"},
"to_time": {"type": "string", "description": "ISO datetime"},
"project": {"type": "string"},
"task": {"type": "string"},
"activity_type": {"type": "string"},
"description": {"type": "string"},
},
},
),
]
def handlers() -> dict:
return {
"erpnext_create_quotation_from_opportunity": _make_quotation_from_opportunity,
"erpnext_create_sales_order_from_quotation": _make_so_from_quotation,
"erpnext_create_sales_invoice_from_sales_order": _make_si_from_so,
"erpnext_create_delivery_note_from_sales_order": _make_dn_from_so,
"erpnext_create_payment_entry_for_invoice": _make_payment_entry,
"erpnext_create_purchase_order_from_supplier_quotation": _make_po_from_sq,
"erpnext_create_purchase_receipt_from_purchase_order": _make_pr_from_po,
"erpnext_create_purchase_invoice_from_purchase_receipt": _make_pi_from_pr,
"erpnext_get_item_stock_balance": _get_item_stock_balance,
"erpnext_transfer_stock": _transfer_stock,
"erpnext_create_stock_entry": _create_stock_entry,
"erpnext_approve_leave_application": _approve_leave,
"erpnext_reject_leave_application": _reject_leave,
"erpnext_mark_attendance": _mark_attendance,
"erpnext_approve_expense_claim": _approve_expense_claim,
"erpnext_create_issue": _create_issue,
"erpnext_close_issue": _close_issue,
"erpnext_create_task": _create_task,
"erpnext_update_task_status": _update_task_status,
"erpnext_log_timesheet_entry": _log_timesheet,
}
async def _make_quotation_from_opportunity(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.crm.doctype.opportunity.opportunity.make_quotation",
source_name=args["opportunity_name"],
)
return json.dumps(result, indent=2)
async def _make_so_from_quotation(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.selling.doctype.quotation.quotation.make_sales_order",
source_name=args["quotation_name"],
)
return json.dumps(result, indent=2)
async def _make_si_from_so(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.selling.doctype.sales_order.sales_order.make_sales_invoice",
source_name=args["sales_order_name"],
)
return json.dumps(result, indent=2)
async def _make_dn_from_so(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.selling.doctype.sales_order.sales_order.make_delivery_note",
source_name=args["sales_order_name"],
)
return json.dumps(result, indent=2)
async def _make_payment_entry(args: dict) -> str:
client = FrappeClient()
method = (
"erpnext.accounts.doctype.sales_invoice.sales_invoice.get_bank_cash_account"
if args["invoice_doctype"] == "Sales Invoice"
else "erpnext.accounts.doctype.purchase_invoice.purchase_invoice.make_payment_entry"
)
result = await client.call_method(
"erpnext.accounts.doctype.payment_entry.payment_entry.get_payment_entry",
dt=args["invoice_doctype"],
dn=args["invoice_name"],
)
return json.dumps(result, indent=2)
async def _make_po_from_sq(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.buying.doctype.supplier_quotation.supplier_quotation.make_purchase_order",
source_name=args["supplier_quotation_name"],
)
return json.dumps(result, indent=2)
async def _make_pr_from_po(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.buying.doctype.purchase_order.purchase_order.make_purchase_receipt",
source_name=args["purchase_order_name"],
)
return json.dumps(result, indent=2)
async def _make_pi_from_pr(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.buying.doctype.purchase_receipt.purchase_receipt.make_purchase_invoice",
source_name=args["purchase_receipt_name"],
)
return json.dumps(result, indent=2)
async def _get_item_stock_balance(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"erpnext.stock.utils.get_stock_balance",
item_code=args["item_code"],
warehouse=args.get("warehouse", ""),
posting_date=args.get("posting_date", ""),
)
return json.dumps({"item_code": args["item_code"], "warehouse": args.get("warehouse"), "balance": result}, indent=2)
async def _transfer_stock(args: dict) -> str:
client = FrappeClient()
payload = {
"stock_entry_type": "Material Transfer",
"posting_date": args.get("posting_date", ""),
"remarks": args.get("remarks", ""),
"items": [{
"item_code": args["item_code"],
"qty": args["qty"],
"s_warehouse": args["from_warehouse"],
"t_warehouse": args["to_warehouse"],
}],
}
result = await client.create_doc("Stock Entry", payload)
return json.dumps(result, indent=2)
async def _create_stock_entry(args: dict) -> str:
client = FrappeClient()
payload = {
"stock_entry_type": args["stock_entry_type"],
"posting_date": args.get("posting_date", ""),
"remarks": args.get("remarks", ""),
"items": args["items"],
}
result = await client.create_doc("Stock Entry", payload)
return json.dumps(result, indent=2)
async def _approve_leave(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("Leave Application", args["leave_application_name"], {"status": "Approved"})
return json.dumps(result, indent=2)
async def _reject_leave(args: dict) -> str:
client = FrappeClient()
updates = {"status": "Rejected"}
if reason := args.get("reason"):
updates["reason_for_rejection"] = reason
result = await client.update_doc("Leave Application", args["leave_application_name"], updates)
return json.dumps(result, indent=2)
async def _mark_attendance(args: dict) -> str:
client = FrappeClient()
payload = {
"employee": args["employee"],
"attendance_date": args["attendance_date"],
"status": args["status"],
"shift": args.get("shift", ""),
"docstatus": 1,
}
result = await client.create_doc("Attendance", payload)
return json.dumps(result, indent=2)
async def _approve_expense_claim(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("Expense Claim", args["expense_claim_name"], {"approval_status": "Approved"})
return json.dumps(result, indent=2)
async def _create_issue(args: dict) -> str:
client = FrappeClient()
payload = {
"subject": args["subject"],
"customer": args.get("customer", ""),
"description": args.get("description", ""),
"priority": args.get("priority", "Medium"),
"issue_type": args.get("issue_type", ""),
"raised_by": args.get("raised_by", ""),
}
result = await client.create_doc("Issue", payload)
return json.dumps(result, indent=2)
async def _close_issue(args: dict) -> str:
client = FrappeClient()
updates = {"status": "Closed"}
if res := args.get("resolution_details"):
updates["resolution_details"] = res
result = await client.update_doc("Issue", args["issue_name"], updates)
return json.dumps(result, indent=2)
async def _create_task(args: dict) -> str:
client = FrappeClient()
result = await client.create_doc("Task", args)
return json.dumps(result, indent=2)
async def _update_task_status(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("Task", args["task_name"], {"status": args["status"]})
return json.dumps(result, indent=2)
async def _log_timesheet(args: dict) -> str:
client = FrappeClient()
payload = {
"employee": args["employee"],
"time_logs": [{
"from_time": args["from_time"],
"to_time": args["to_time"],
"project": args.get("project", ""),
"task": args.get("task", ""),
"activity_type": args.get("activity_type", ""),
"description": args.get("description", ""),
}],
}
result = await client.create_doc("Timesheet", payload)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,124 @@
"""Custom Fields tools — add, modify, remove custom fields on any DocType."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
_FIELD_TYPES = [
"Data", "Int", "Float", "Currency", "Percent", "Check",
"Small Text", "Text", "Long Text", "Text Editor", "HTML",
"Date", "Datetime", "Time", "Duration",
"Select", "Link", "Dynamic Link", "Table", "Table MultiSelect",
"Attach", "Attach Image", "Signature", "Color", "Barcode",
"Geolocation", "Rating", "Section Break", "Column Break", "Tab Break",
]
_FIELD_SCHEMA = {
"type": "object",
"required": ["dt", "fieldname", "fieldtype", "label"],
"properties": {
"dt": {"type": "string", "description": "DocType to add the field to"},
"fieldname": {"type": "string"},
"fieldtype": {"type": "string", "enum": _FIELD_TYPES},
"label": {"type": "string"},
"options": {"type": "string", "description": "For Link: target DocType. For Select: newline-separated options."},
"reqd": {"type": "integer", "default": 0},
"in_list_view": {"type": "integer", "default": 0},
"in_standard_filter": {"type": "integer", "default": 0},
"insert_after": {"type": "string", "description": "Fieldname to insert after"},
"default": {"type": "string"},
"description": {"type": "string"},
"hidden": {"type": "integer", "default": 0},
"read_only": {"type": "integer", "default": 0},
"bold": {"type": "integer", "default": 0},
},
}
def tools() -> list[Tool]:
return [
Tool(
name="frappe_add_custom_field",
description="Add a custom field to an existing DocType.",
inputSchema=_FIELD_SCHEMA,
),
Tool(
name="frappe_update_custom_field",
description="Update properties of an existing custom field.",
inputSchema={
"type": "object",
"required": ["dt", "fieldname"],
"properties": {
"dt": {"type": "string"},
"fieldname": {"type": "string"},
"updates": {"type": "object", "description": "Properties to update"},
},
},
),
Tool(
name="frappe_remove_custom_field",
description="Remove a custom field from a DocType.",
inputSchema={
"type": "object",
"required": ["dt", "fieldname"],
"properties": {
"dt": {"type": "string"},
"fieldname": {"type": "string"},
},
},
),
Tool(
name="frappe_list_custom_fields",
description="List all custom fields for a DocType.",
inputSchema={
"type": "object",
"required": ["dt"],
"properties": {
"dt": {"type": "string"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_add_custom_field": _add_custom_field,
"frappe_update_custom_field": _update_custom_field,
"frappe_remove_custom_field": _remove_custom_field,
"frappe_list_custom_fields": _list_custom_fields,
}
async def _add_custom_field(args: dict) -> str:
client = FrappeClient()
payload = {k: v for k, v in args.items()}
result = await client.create_doc("Custom Field", payload)
return json.dumps(result, indent=2)
async def _update_custom_field(args: dict) -> str:
client = FrappeClient()
# Custom Field name is "{DocType}-{fieldname}"
cf_name = f"{args['dt']}-{args['fieldname']}"
result = await client.update_doc("Custom Field", cf_name, args.get("updates", {}))
return json.dumps(result, indent=2)
async def _remove_custom_field(args: dict) -> str:
client = FrappeClient()
cf_name = f"{args['dt']}-{args['fieldname']}"
result = await client.delete_doc("Custom Field", cf_name)
return json.dumps(result, indent=2)
async def _list_custom_fields(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Custom Field",
fields=["name", "fieldname", "fieldtype", "label", "reqd", "hidden"],
filters=[["dt", "=", args["dt"]]],
limit=100,
)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,217 @@
"""Dashboard, Chart, Number Card, and Workspace tools."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_list_dashboards",
description="List all Frappe Dashboards.",
inputSchema={
"type": "object",
"properties": {"limit": {"type": "integer", "default": 20}},
},
),
Tool(
name="frappe_get_dashboard",
description="Get a Dashboard with all its charts and cards.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {"name": {"type": "string"}},
},
),
Tool(
name="frappe_create_dashboard",
description="Create a new Dashboard.",
inputSchema={
"type": "object",
"required": ["dashboard_name"],
"properties": {
"dashboard_name": {"type": "string"},
"module": {"type": "string"},
"is_default": {"type": "integer", "default": 0},
"charts": {
"type": "array",
"items": {"type": "object"},
"description": "List of {chart, width} objects",
},
},
},
),
Tool(
name="frappe_create_dashboard_chart",
description="Create a Dashboard Chart (bar, line, pie, donut, percentage, heatmap).",
inputSchema={
"type": "object",
"required": ["chart_name", "chart_type", "document_type"],
"properties": {
"chart_name": {"type": "string"},
"chart_type": {
"type": "string",
"enum": ["Count", "Sum", "Average", "Min", "Max", "Group By"],
},
"document_type": {"type": "string"},
"based_on": {"type": "string", "description": "Date field for time-series grouping"},
"value_based_on": {"type": "string", "description": "Field to aggregate"},
"type": {
"type": "string",
"enum": ["Bar", "Line", "Percentage", "Pie", "Donut", "Heatmap"],
"default": "Bar",
},
"timespan": {
"type": "string",
"enum": ["Last Year", "Last Quarter", "Last Month", "Last Week"],
"default": "Last Year",
},
"time_interval": {
"type": "string",
"enum": ["Yearly", "Quarterly", "Monthly", "Weekly", "Daily"],
"default": "Monthly",
},
"filters_json": {"type": "string", "description": "JSON string of filters"},
"color": {"type": "string"},
},
},
),
Tool(
name="frappe_create_number_card",
description="Create a Number Card for dashboards showing a single aggregated metric.",
inputSchema={
"type": "object",
"required": ["name", "document_type", "function"],
"properties": {
"name": {"type": "string"},
"label": {"type": "string"},
"document_type": {"type": "string"},
"function": {
"type": "string",
"enum": ["Count", "Sum", "Average", "Min", "Max"],
"default": "Count",
},
"aggregate_function_based_on": {"type": "string"},
"filters_json": {"type": "string"},
"color": {"type": "string"},
"stats_time_interval": {"type": "string", "enum": ["Daily", "Weekly", "Monthly", "Quarterly", "Yearly"]},
},
},
),
# --- Workspace ---
Tool(
name="frappe_list_workspaces",
description="List all Desk Workspaces.",
inputSchema={
"type": "object",
"properties": {"limit": {"type": "integer", "default": 30}},
},
),
Tool(
name="frappe_get_workspace",
description="Get a Workspace with all its shortcuts, links, and charts.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {"name": {"type": "string"}},
},
),
Tool(
name="frappe_create_workspace",
description="Create a new Desk Workspace with shortcuts and links.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
"label": {"type": "string"},
"module": {"type": "string"},
"icon": {"type": "string"},
"is_hidden": {"type": "integer", "default": 0},
"shortcuts": {
"type": "array",
"items": {"type": "object"},
"description": "List of shortcut objects {label, link_to, type}",
},
"links": {
"type": "array",
"items": {"type": "object"},
"description": "List of link objects for the workspace body",
},
},
},
),
Tool(
name="frappe_update_workspace",
description="Update a Workspace's content, shortcuts, or visibility.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
"updates": {"type": "object"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_list_dashboards": _list_dashboards,
"frappe_get_dashboard": _get_dashboard,
"frappe_create_dashboard": _create_dashboard,
"frappe_create_dashboard_chart": _create_dashboard_chart,
"frappe_create_number_card": _create_number_card,
"frappe_list_workspaces": _list_workspaces,
"frappe_get_workspace": _get_workspace,
"frappe_create_workspace": _create_workspace,
"frappe_update_workspace": _update_workspace,
}
async def _list_dashboards(args: dict) -> str:
client = FrappeClient()
result = await client.get_list("Dashboard", fields=["name", "module", "is_default"], limit=args.get("limit", 20))
return json.dumps(result, indent=2)
async def _get_dashboard(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("Dashboard", args["name"])
return json.dumps(result, indent=2)
async def _create_dashboard(args: dict) -> str:
client = FrappeClient()
result = await client.create_doc("Dashboard", args)
return json.dumps(result, indent=2)
async def _create_dashboard_chart(args: dict) -> str:
client = FrappeClient()
result = await client.create_doc("Dashboard Chart", args)
return json.dumps(result, indent=2)
async def _create_number_card(args: dict) -> str:
client = FrappeClient()
result = await client.create_doc("Number Card", args)
return json.dumps(result, indent=2)
async def _list_workspaces(args: dict) -> str:
client = FrappeClient()
result = await client.get_list("Workspace", fields=["name", "label", "module", "is_hidden"], limit=args.get("limit", 30))
return json.dumps(result, indent=2)
async def _get_workspace(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("Workspace", args["name"])
return json.dumps(result, indent=2)
async def _create_workspace(args: dict) -> str:
client = FrappeClient()
result = await client.create_doc("Workspace", args)
return json.dumps(result, indent=2)
async def _update_workspace(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("Workspace", args["name"], args.get("updates", {}))
return json.dumps(result, indent=2)

View File

@ -0,0 +1,155 @@
"""DocType management tools — create, inspect, list, and modify DocTypes."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_list_doctypes",
description="List all DocTypes in the Frappe instance, optionally filtered by module.",
inputSchema={
"type": "object",
"properties": {
"module": {"type": "string", "description": "Filter by module name (optional)"},
"limit": {"type": "integer", "default": 50},
},
},
),
Tool(
name="frappe_get_doctype",
description="Get the full schema/definition of a DocType including all fields.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string", "description": "DocType name e.g. 'Sales Order'"},
},
},
),
Tool(
name="frappe_create_doctype",
description=(
"Create a new custom DocType. Provide name, module, and a list of fields. "
"Each field needs at minimum: fieldname, fieldtype, label."
),
inputSchema={
"type": "object",
"required": ["name", "module", "fields"],
"properties": {
"name": {"type": "string"},
"module": {"type": "string"},
"is_submittable": {"type": "boolean", "default": False},
"istable": {"type": "integer", "default": 0, "description": "Set 1 to make this a child table DocType"},
"track_changes": {"type": "boolean", "default": True},
"autoname": {"type": "string", "description": "e.g. 'hash', 'field:fieldname', 'naming_series:'"},
"title_field": {"type": "string", "description": "Field to use as document title"},
"search_fields": {"type": "string", "description": "Comma-separated fields for search"},
"sort_field": {"type": "string"},
"sort_order": {"type": "string", "enum": ["ASC", "DESC"], "default": "DESC"},
"fields": {
"type": "array",
"items": {
"type": "object",
"required": ["fieldname", "fieldtype", "label"],
"properties": {
"fieldname": {"type": "string"},
"fieldtype": {"type": "string"},
"label": {"type": "string"},
"reqd": {"type": "integer", "default": 0},
"in_list_view": {"type": "integer", "default": 0},
"options": {"type": "string"},
},
},
},
"permissions": {
"type": "array",
"description": "List of permission rows. Defaults to System Manager full access.",
},
},
},
),
Tool(
name="frappe_update_doctype",
description="Update an existing DocType — add/remove fields or change properties.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"updates": {
"type": "object",
"description": "Fields to update on the DocType document (e.g. {\"fields\": [...]})",
},
},
},
),
]
def handlers() -> dict:
return {
"frappe_list_doctypes": _list_doctypes,
"frappe_get_doctype": _get_doctype,
"frappe_create_doctype": _create_doctype,
"frappe_update_doctype": _update_doctype,
}
async def _list_doctypes(args: dict) -> str:
client = FrappeClient()
filters = []
if module := args.get("module"):
filters.append(["module", "=", module])
result = await client.get_list(
"DocType",
fields=["name", "module", "is_single", "is_submittable", "modified"],
filters=filters if filters else None,
limit=args.get("limit", 50),
)
return json.dumps(result, indent=2)
async def _get_doctype(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("DocType", args["doctype"])
return json.dumps(result, indent=2)
async def _create_doctype(args: dict) -> str:
client = FrappeClient()
istable = int(args.get("istable", 0))
payload = {
"name": args["name"],
"module": args["module"],
"custom": 1,
"istable": istable,
"is_submittable": int(args.get("is_submittable", False)),
"track_changes": int(args.get("track_changes", True)),
"fields": args["fields"],
}
if v := args.get("autoname"):
payload["autoname"] = v
if v := args.get("title_field"):
payload["title_field"] = v
if v := args.get("search_fields"):
payload["search_fields"] = v
if v := args.get("sort_field"):
payload["sort_field"] = v
if v := args.get("sort_order"):
payload["sort_order"] = v
# child tables don't need permissions
if not istable:
payload["permissions"] = args.get("permissions", [
{"role": "System Manager", "read": 1, "write": 1, "create": 1, "delete": 1, "submit": 0},
])
result = await client.create_doc("DocType", payload)
return json.dumps(result, indent=2)
async def _update_doctype(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("DocType", args["doctype"], args.get("updates", {}))
return json.dumps(result, indent=2)

View File

@ -0,0 +1,267 @@
"""
Level 2 Document inspection tools.
Search, linked records, timeline, children, counts.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_search_documents",
description=(
"Generic full-text search across any DocType. "
"Searches title/name fields. Use filters for precise queries."
),
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"query": {"type": "string", "description": "Search text"},
"filters": {"type": "array", "description": "Additional filter conditions"},
"fields": {"type": "array", "items": {"type": "string"}},
"limit": {"type": "integer", "default": 20},
"order_by": {"type": "string", "default": "modified desc"},
},
},
),
Tool(
name="frappe_get_document_children",
description=(
"Read child table rows for a document. "
"Use get_doctype_meta to find the child table fieldname first."
),
inputSchema={
"type": "object",
"required": ["doctype", "name", "child_doctype"],
"properties": {
"doctype": {"type": "string", "description": "Parent DocType"},
"name": {"type": "string", "description": "Parent document name"},
"child_doctype": {"type": "string", "description": "Child table DocType e.g. 'Sales Order Item'"},
"fields": {"type": "array", "items": {"type": "string"}},
},
},
),
Tool(
name="frappe_get_linked_documents",
description=(
"Returns all documents that link TO this document — "
"e.g. all Sales Orders linked to a Customer, or all Invoices for a Sales Order."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"linked_doctype": {
"type": "string",
"description": "Filter to only this DocType (optional)",
},
"limit": {"type": "integer", "default": 20},
},
},
),
Tool(
name="frappe_get_document_timeline",
description=(
"Get the full activity timeline for a document: "
"comments, emails, assignments, workflow changes, version history."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"limit": {"type": "integer", "default": 30},
},
},
),
Tool(
name="frappe_get_recent_documents",
description="Get recently modified documents of a DocType. Useful for 'show latest X' requests.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"fields": {"type": "array", "items": {"type": "string"}},
"limit": {"type": "integer", "default": 10},
"modified_after": {"type": "string", "description": "ISO datetime filter e.g. '2024-01-01'"},
},
},
),
Tool(
name="frappe_count_documents",
description="Count records matching filters. Fast — does not return document data.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"filters": {"type": "array"},
},
},
),
Tool(
name="frappe_get_document_attachments",
description="List all files attached to a specific document.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_get_document_versions",
description="Get the change history (versions) of a document — who changed what fields and when.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"limit": {"type": "integer", "default": 10},
},
},
),
]
def handlers() -> dict:
return {
"frappe_search_documents": _search_documents,
"frappe_get_document_children": _get_document_children,
"frappe_get_linked_documents": _get_linked_documents,
"frappe_get_document_timeline": _get_document_timeline,
"frappe_get_recent_documents": _get_recent_documents,
"frappe_count_documents": _count_documents,
"frappe_get_document_attachments": _get_document_attachments,
"frappe_get_document_versions": _get_document_versions,
}
async def _search_documents(args: dict) -> str:
client = FrappeClient()
filters = list(args.get("filters") or [])
if query := args.get("query"):
filters.append(["name", "like", f"%{query}%"])
result = await client.get_list(
args["doctype"],
fields=args.get("fields", ["name", "modified", "owner"]),
filters=filters if filters else None,
limit=args.get("limit", 20),
order_by=args.get("order_by", "modified desc"),
)
return json.dumps(result, indent=2)
async def _get_document_children(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
args["child_doctype"],
fields=args.get("fields", ["*"]) or ["*"],
filters=[["parent", "=", args["name"]], ["parenttype", "=", args["doctype"]]],
limit=500,
order_by="idx asc",
)
return json.dumps(result, indent=2)
async def _get_linked_documents(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.client.get_linked_docs",
doctype=args["doctype"],
name=args["name"],
linkinfo=args.get("linked_doctype", ""),
)
return json.dumps(result, indent=2)
async def _get_document_timeline(args: dict) -> str:
client = FrappeClient()
ref_dt = args["doctype"]
ref_name = args["name"]
limit = args.get("limit", 30)
comments = await client.get_list(
"Comment",
fields=["name", "comment_type", "comment_by", "content", "creation"],
filters=[["reference_doctype", "=", ref_dt], ["reference_name", "=", ref_name]],
limit=limit,
order_by="creation desc",
)
versions = await client.get_list(
"Version",
fields=["name", "owner", "creation", "data"],
filters=[["ref_doctype", "=", ref_dt], ["docname", "=", ref_name]],
limit=limit,
order_by="creation desc",
)
timeline = sorted(
[{"type": "comment", **c} for c in (comments if isinstance(comments, list) else [])] +
[{"type": "version", **v} for v in (versions if isinstance(versions, list) else [])],
key=lambda x: x.get("creation", ""),
reverse=True,
)[:limit]
return json.dumps(timeline, indent=2)
async def _get_recent_documents(args: dict) -> str:
client = FrappeClient()
filters = []
if after := args.get("modified_after"):
filters.append(["modified", ">", after])
result = await client.get_list(
args["doctype"],
fields=args.get("fields", ["name", "modified", "owner", "docstatus"]),
filters=filters if filters else None,
limit=args.get("limit", 10),
order_by="modified desc",
)
return json.dumps(result, indent=2)
async def _count_documents(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.client.get_count",
doctype=args["doctype"],
filters=args.get("filters"),
)
return json.dumps({"doctype": args["doctype"], "count": result}, indent=2)
async def _get_document_attachments(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"File",
fields=["name", "file_name", "file_url", "file_size", "is_private", "creation"],
filters=[
["attached_to_doctype", "=", args["doctype"]],
["attached_to_name", "=", args["name"]],
],
limit=50,
)
return json.dumps(result, indent=2)
async def _get_document_versions(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Version",
fields=["name", "owner", "creation", "data"],
filters=[["ref_doctype", "=", args["doctype"]], ["docname", "=", args["name"]]],
limit=args.get("limit", 10),
order_by="creation desc",
)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,229 @@
"""
Level 3-4 Document lifecycle tools.
Child row operations, rename, amend, duplicate, status, draft save.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
# --- Child row ops ---
Tool(
name="frappe_append_child_row",
description="Append a new row to a child table in a document.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "child_fieldname", "row_data"],
"properties": {
"doctype": {"type": "string", "description": "Parent DocType"},
"name": {"type": "string", "description": "Parent document name"},
"child_fieldname": {"type": "string", "description": "Fieldname of the child table e.g. 'items'"},
"row_data": {"type": "object", "description": "Fields for the new row"},
},
},
),
Tool(
name="frappe_update_child_row",
description="Update a specific row in a child table by its row name.",
inputSchema={
"type": "object",
"required": ["child_doctype", "row_name", "updates"],
"properties": {
"child_doctype": {"type": "string", "description": "Child table DocType e.g. 'Sales Order Item'"},
"row_name": {"type": "string", "description": "Child row document name"},
"updates": {"type": "object"},
},
},
),
Tool(
name="frappe_delete_child_row",
description="Delete a specific row from a child table.",
inputSchema={
"type": "object",
"required": ["child_doctype", "row_name"],
"properties": {
"child_doctype": {"type": "string"},
"row_name": {"type": "string"},
},
},
),
# --- Rename ---
Tool(
name="frappe_rename_document",
description="Rename a document (changes its primary key/name). Only works if DocType allows renaming.",
inputSchema={
"type": "object",
"required": ["doctype", "old_name", "new_name"],
"properties": {
"doctype": {"type": "string"},
"old_name": {"type": "string"},
"new_name": {"type": "string"},
"merge": {"type": "boolean", "default": False, "description": "Merge into existing document if new_name exists"},
},
},
),
# --- Amend ---
Tool(
name="frappe_amend_document",
description=(
"Create an amendment of a cancelled/submitted document. "
"Returns a new draft with amended_ prefix and links to original."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
# --- Duplicate ---
Tool(
name="frappe_duplicate_document",
description="Clone a document — creates a new draft copy with the same field values.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"field_overrides": {
"type": "object",
"description": "Fields to override in the copy e.g. {\"customer\": \"New Customer\"}",
},
},
},
),
# --- Status ---
Tool(
name="frappe_get_document_status",
description=(
"Get the normalized status of a document: "
"draft | saved | submitted | cancelled | workflow_state."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
# --- Draft save ---
Tool(
name="frappe_save_document_draft",
description=(
"Explicitly save a document as draft (docstatus=0). "
"Use when you want to save partial data without validating all required fields."
),
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"updates": {"type": "object", "description": "Optional field updates to apply before saving"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_append_child_row": _append_child_row,
"frappe_update_child_row": _update_child_row,
"frappe_delete_child_row": _delete_child_row,
"frappe_rename_document": _rename_document,
"frappe_amend_document": _amend_document,
"frappe_duplicate_document": _duplicate_document,
"frappe_get_document_status": _get_document_status,
"frappe_save_document_draft": _save_document_draft,
}
_DOCSTATUS_MAP = {0: "draft/saved", 1: "submitted", 2: "cancelled"}
async def _append_child_row(args: dict) -> str:
client = FrappeClient()
doc = await client.get_doc(args["doctype"], args["name"])
rows = doc.get(args["child_fieldname"], [])
rows.append(args["row_data"])
result = await client.update_doc(args["doctype"], args["name"], {args["child_fieldname"]: rows})
return json.dumps(result, indent=2)
async def _update_child_row(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc(args["child_doctype"], args["row_name"], args["updates"])
return json.dumps(result, indent=2)
async def _delete_child_row(args: dict) -> str:
client = FrappeClient()
result = await client.delete_doc(args["child_doctype"], args["row_name"])
return json.dumps(result, indent=2)
async def _rename_document(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.client.rename_doc",
doctype=args["doctype"],
old_name=args["old_name"],
new_name=args["new_name"],
merge=args.get("merge", False),
)
return json.dumps({"renamed_to": result}, indent=2)
async def _amend_document(args: dict) -> str:
client = FrappeClient()
doc = await client.get_doc(args["doctype"], args["name"])
amended = {k: v for k, v in doc.items() if not k.startswith("__")}
amended.pop("name", None)
amended["amended_from"] = args["name"]
amended["docstatus"] = 0
result = await client.create_doc(args["doctype"], amended)
return json.dumps(result, indent=2)
async def _duplicate_document(args: dict) -> str:
client = FrappeClient()
doc = await client.get_doc(args["doctype"], args["name"])
copy = {k: v for k, v in doc.items() if not k.startswith("__")}
copy.pop("name", None)
copy["docstatus"] = 0
copy.update(args.get("field_overrides", {}))
result = await client.create_doc(args["doctype"], copy)
return json.dumps(result, indent=2)
async def _get_document_status(args: dict) -> str:
client = FrappeClient()
doc = await client.get_doc(args["doctype"], args["name"])
docstatus = doc.get("docstatus", 0)
workflow_state = doc.get("workflow_state", None)
status = doc.get("status", None)
return json.dumps({
"name": args["name"],
"doctype": args["doctype"],
"docstatus": docstatus,
"docstatus_label": _DOCSTATUS_MAP.get(docstatus, "unknown"),
"status": status,
"workflow_state": workflow_state,
}, indent=2)
async def _save_document_draft(args: dict) -> str:
client = FrappeClient()
updates = args.get("updates", {})
updates["docstatus"] = 0
result = await client.update_doc(args["doctype"], args["name"], updates)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,169 @@
"""Document CRUD tools — create, read, update, delete, list Frappe documents."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
from frappe_mcp.config import get_settings
def tools() -> list[Tool]:
return [
Tool(
name="frappe_list_documents",
description="List documents of a given DocType with optional filters.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"fields": {
"type": "array",
"items": {"type": "string"},
"description": "Fields to return. Defaults to ['name', 'modified'].",
},
"filters": {
"type": "array",
"description": "Filter list e.g. [[\"status\", \"=\", \"Open\"]]",
},
"limit": {"type": "integer", "default": 20},
"order_by": {"type": "string", "default": "modified desc"},
},
},
),
Tool(
name="frappe_get_document",
description="Get a single Frappe document by DocType and name.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_create_document",
description="Create a new document in any DocType.",
inputSchema={
"type": "object",
"required": ["doctype", "data"],
"properties": {
"doctype": {"type": "string"},
"data": {"type": "object", "description": "Field values for the new document"},
},
},
),
Tool(
name="frappe_update_document",
description="Update fields on an existing document.",
inputSchema={
"type": "object",
"required": ["doctype", "name", "data"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
"data": {"type": "object", "description": "Fields to update"},
},
},
),
Tool(
name="frappe_delete_document",
description="Delete a document. Blocked in read-only mode.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_submit_document",
description="Submit a submittable document (changes status to Submitted).",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_cancel_document",
description="Cancel a submitted document.",
inputSchema={
"type": "object",
"required": ["doctype", "name"],
"properties": {
"doctype": {"type": "string"},
"name": {"type": "string"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_list_documents": _list_documents,
"frappe_get_document": _get_document,
"frappe_create_document": _create_document,
"frappe_update_document": _update_document,
"frappe_delete_document": _delete_document,
"frappe_submit_document": _submit_document,
"frappe_cancel_document": _cancel_document,
}
async def _list_documents(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
args["doctype"],
fields=args.get("fields", ["name", "modified"]),
filters=args.get("filters"),
limit=args.get("limit", 20),
order_by=args.get("order_by", "modified desc"),
)
return json.dumps(result, indent=2)
async def _get_document(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc(args["doctype"], args["name"])
return json.dumps(result, indent=2)
async def _create_document(args: dict) -> str:
client = FrappeClient()
data = {"doctype": args["doctype"], **args["data"]}
result = await client.create_doc(args["doctype"], data)
return json.dumps(result, indent=2)
async def _update_document(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc(args["doctype"], args["name"], args["data"])
return json.dumps(result, indent=2)
async def _delete_document(args: dict) -> str:
if get_settings().read_only_mode:
return "Error: read_only_mode is enabled. Deletion blocked."
client = FrappeClient()
result = await client.delete_doc(args["doctype"], args["name"])
return json.dumps(result, indent=2)
async def _submit_document(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc(args["doctype"], args["name"], {"docstatus": 1})
return json.dumps(result, indent=2)
async def _cancel_document(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc(args["doctype"], args["name"], {"docstatus": 2})
return json.dumps(result, indent=2)

View File

@ -0,0 +1,220 @@
"""Email Template tools — create, list, update Frappe Email Templates and Notifications."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_list_email_templates",
description="List all Email Templates in the system.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "default": 30},
},
},
),
Tool(
name="frappe_get_email_template",
description="Get a specific Email Template including subject and body.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_create_email_template",
description=(
"Create a new Email Template. Subject and response support Jinja2. "
"Access doc fields with {{ doc.field_name }}."
),
inputSchema={
"type": "object",
"required": ["name", "subject", "response"],
"properties": {
"name": {"type": "string", "description": "Template name/ID"},
"subject": {"type": "string", "description": "Email subject (Jinja2 supported)"},
"response": {"type": "string", "description": "Email body HTML (Jinja2 supported)"},
"use_html": {"type": "integer", "default": 1},
},
},
),
Tool(
name="frappe_update_email_template",
description="Update an Email Template's subject or body.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
"subject": {"type": "string"},
"response": {"type": "string"},
"updates": {"type": "object"},
},
},
),
Tool(
name="frappe_list_notifications",
description="List Frappe Notifications (automated email/Slack alerts on DocType events).",
inputSchema={
"type": "object",
"properties": {
"document_type": {"type": "string", "description": "Filter by DocType"},
"enabled": {"type": "integer", "description": "1 = enabled only"},
"limit": {"type": "integer", "default": 30},
},
},
),
Tool(
name="frappe_create_notification",
description=(
"Create a Frappe Notification — triggers an email/Slack/system message "
"on a DocType event or date condition."
),
inputSchema={
"type": "object",
"required": ["name", "document_type", "event", "subject", "message"],
"properties": {
"name": {"type": "string"},
"document_type": {"type": "string"},
"event": {
"type": "string",
"enum": [
"New", "Save", "Submit", "Cancel", "Days Before",
"Days After", "Value Change", "Method", "Custom",
],
},
"subject": {"type": "string", "description": "Email subject (Jinja2)"},
"message": {"type": "string", "description": "Email body (Jinja2 HTML)"},
"send_to_all_assignees": {"type": "integer", "default": 0},
"recipients": {
"type": "array",
"description": "List of recipient rows: [{\"receiver_by_document_field\": \"email\"}]",
"items": {"type": "object"},
},
"condition": {"type": "string", "description": "Python expression e.g. doc.status == 'Open'"},
"channel": {
"type": "string",
"enum": ["Email", "Slack", "System Notification", "SMS"],
"default": "Email",
},
"enabled": {"type": "integer", "default": 1},
},
},
),
Tool(
name="frappe_update_notification",
description="Enable, disable, or update a Notification.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
"enabled": {"type": "integer"},
"subject": {"type": "string"},
"message": {"type": "string"},
"updates": {"type": "object"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_list_email_templates": _list_email_templates,
"frappe_get_email_template": _get_email_template,
"frappe_create_email_template": _create_email_template,
"frappe_update_email_template": _update_email_template,
"frappe_list_notifications": _list_notifications,
"frappe_create_notification": _create_notification,
"frappe_update_notification": _update_notification,
}
async def _list_email_templates(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Email Template",
fields=["name", "subject", "modified"],
limit=args.get("limit", 30),
)
return json.dumps(result, indent=2)
async def _get_email_template(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("Email Template", args["name"])
return json.dumps(result, indent=2)
async def _create_email_template(args: dict) -> str:
client = FrappeClient()
payload = {
"name": args["name"],
"subject": args["subject"],
"response": args["response"],
"use_html": args.get("use_html", 1),
}
result = await client.create_doc("Email Template", payload)
return json.dumps(result, indent=2)
async def _update_email_template(args: dict) -> str:
client = FrappeClient()
updates = args.get("updates", {})
for field in ("subject", "response"):
if field in args:
updates[field] = args[field]
result = await client.update_doc("Email Template", args["name"], updates)
return json.dumps(result, indent=2)
async def _list_notifications(args: dict) -> str:
client = FrappeClient()
filters = []
if dt := args.get("document_type"):
filters.append(["document_type", "=", dt])
if "enabled" in args:
filters.append(["enabled", "=", args["enabled"]])
result = await client.get_list(
"Notification",
fields=["name", "document_type", "event", "channel", "enabled"],
filters=filters if filters else None,
limit=args.get("limit", 30),
)
return json.dumps(result, indent=2)
async def _create_notification(args: dict) -> str:
client = FrappeClient()
payload = {
"name": args["name"],
"document_type": args["document_type"],
"event": args["event"],
"subject": args["subject"],
"message": args["message"],
"channel": args.get("channel", "Email"),
"enabled": args.get("enabled", 1),
"send_to_all_assignees": args.get("send_to_all_assignees", 0),
"recipients": args.get("recipients", []),
"condition": args.get("condition", ""),
}
result = await client.create_doc("Notification", payload)
return json.dumps(result, indent=2)
async def _update_notification(args: dict) -> str:
client = FrappeClient()
updates = args.get("updates", {})
for field in ("enabled", "subject", "message"):
if field in args:
updates[field] = args[field]
result = await client.update_doc("Notification", args["name"], updates)
return json.dumps(result, indent=2)

168
frappe_mcp/tools/files.py Normal file
View File

@ -0,0 +1,168 @@
"""File and attachment management tools."""
import json
import base64
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_list_files",
description="List files in Frappe's file manager, optionally filtered by attached document.",
inputSchema={
"type": "object",
"properties": {
"attached_to_doctype": {"type": "string"},
"attached_to_name": {"type": "string"},
"folder": {"type": "string", "description": "e.g. 'Home/Attachments'"},
"is_private": {"type": "integer", "description": "1 = private, 0 = public"},
"limit": {"type": "integer", "default": 20},
},
},
),
Tool(
name="frappe_get_file",
description="Get details of a file by its name/ID.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string"},
},
},
),
Tool(
name="frappe_upload_file_base64",
description=(
"Upload a file to Frappe by providing base64-encoded content. "
"Optionally attach it to a document."
),
inputSchema={
"type": "object",
"required": ["filename", "content_base64"],
"properties": {
"filename": {"type": "string", "description": "File name with extension"},
"content_base64": {"type": "string", "description": "Base64-encoded file content"},
"attached_to_doctype": {"type": "string"},
"attached_to_name": {"type": "string"},
"attached_to_field": {"type": "string"},
"is_private": {"type": "integer", "default": 1},
"folder": {"type": "string", "default": "Home/Attachments"},
},
},
),
Tool(
name="frappe_delete_file",
description="Delete a file from Frappe.",
inputSchema={
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string", "description": "File document name"},
},
},
),
Tool(
name="frappe_move_file",
description="Move a file to a different folder.",
inputSchema={
"type": "object",
"required": ["name", "folder"],
"properties": {
"name": {"type": "string"},
"folder": {"type": "string", "description": "Target folder path"},
},
},
),
Tool(
name="frappe_create_folder",
description="Create a new folder in the Frappe file manager.",
inputSchema={
"type": "object",
"required": ["folder_name"],
"properties": {
"folder_name": {"type": "string"},
"parent_folder": {"type": "string", "default": "Home"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_list_files": _list_files,
"frappe_get_file": _get_file,
"frappe_upload_file_base64": _upload_file_base64,
"frappe_delete_file": _delete_file,
"frappe_move_file": _move_file,
"frappe_create_folder": _create_folder,
}
async def _list_files(args: dict) -> str:
client = FrappeClient()
filters = []
if dt := args.get("attached_to_doctype"):
filters.append(["attached_to_doctype", "=", dt])
if name := args.get("attached_to_name"):
filters.append(["attached_to_name", "=", name])
if folder := args.get("folder"):
filters.append(["folder", "=", folder])
if "is_private" in args:
filters.append(["is_private", "=", args["is_private"]])
result = await client.get_list(
"File",
fields=["name", "file_name", "file_url", "file_size", "is_private", "folder", "attached_to_doctype", "attached_to_name"],
filters=filters if filters else None,
limit=args.get("limit", 20),
)
return json.dumps(result, indent=2)
async def _get_file(args: dict) -> str:
client = FrappeClient()
result = await client.get_doc("File", args["name"])
return json.dumps(result, indent=2)
async def _upload_file_base64(args: dict) -> str:
client = FrappeClient()
payload = {
"filename": args["filename"],
"filedata": args["content_base64"],
"is_private": args.get("is_private", 1),
"folder": args.get("folder", "Home/Attachments"),
}
if dt := args.get("attached_to_doctype"):
payload["doctype"] = dt
if name := args.get("attached_to_name"):
payload["docname"] = name
if field := args.get("attached_to_field"):
payload["fieldname"] = field
result = await client.call_method("frappe.client.attach_file", **payload)
return json.dumps(result, indent=2)
async def _delete_file(args: dict) -> str:
client = FrappeClient()
result = await client.delete_doc("File", args["name"])
return json.dumps(result, indent=2)
async def _move_file(args: dict) -> str:
client = FrappeClient()
result = await client.update_doc("File", args["name"], {"folder": args["folder"]})
return json.dumps(result, indent=2)
async def _create_folder(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.core.doctype.file.file.create_new_folder",
file_name=args["folder_name"],
folder=args.get("parent_folder", "Home"),
)
return json.dumps(result, indent=2)

View File

@ -0,0 +1,144 @@
"""
Level 1 Foundation tools.
These are the absolute minimum. Without these, the AI is blind.
"""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_ping",
description="Check whether Frappe is reachable. Returns version and status.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_get_session_info",
description=(
"Returns the current session: logged-in user, roles, site name, "
"enabled modules, and Frappe version. Call this first to understand context."
),
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_get_doctype_permissions",
description=(
"Returns what the current user can do on a DocType: "
"read, write, create, delete, submit, cancel, amend, print, email, export, import."
),
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
},
},
),
Tool(
name="frappe_list_modules",
description=(
"List all Frappe/ERPNext modules available on the site "
"(Selling, Buying, Stock, Accounts, HR, CRM, Support, Custom, etc.)."
),
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_get_doctype_meta",
description=(
"Get comprehensive DocType schema: all fields with types, required flags, "
"options, child tables, link fields, and permission hints. "
"Essential before creating or editing documents."
),
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
"include_permissions": {"type": "boolean", "default": True},
},
},
),
]
def handlers() -> dict:
return {
"frappe_ping": _ping,
"frappe_get_session_info": _get_session_info,
"frappe_get_doctype_permissions": _get_doctype_permissions,
"frappe_list_modules": _list_modules,
"frappe_get_doctype_meta": _get_doctype_meta,
}
async def _ping(args: dict) -> str:
client = FrappeClient()
try:
versions = await client.call_method("frappe.utils.change_log.get_versions")
frappe_ver = versions.get("frappe", {}).get("version", "unknown") if isinstance(versions, dict) else "unknown"
return json.dumps({"status": "ok", "frappe_version": frappe_ver, "url": client.base_url}, indent=2)
except Exception as e:
return json.dumps({"status": "unreachable", "error": str(e)}, indent=2)
async def _get_session_info(args: dict) -> str:
client = FrappeClient()
user = await client.call_method("frappe.auth.get_logged_user")
user_doc = await client.get_doc("User", user)
roles = [r["role"] for r in user_doc.get("roles", [])]
versions = await client.call_method("frappe.utils.change_log.get_versions")
modules_raw = await client.get_list(
"Module Def",
fields=["name", "app_name"],
limit=100,
)
return json.dumps({
"user": user,
"roles": roles,
"frappe_url": client.base_url,
"site_name": client.settings.frappe_site_name or "(default)",
"frappe_version": versions.get("frappe", {}).get("version") if isinstance(versions, dict) else None,
"installed_apps": list(versions.keys()) if isinstance(versions, dict) else [],
"modules": [m["name"] for m in (modules_raw if isinstance(modules_raw, list) else [])],
}, indent=2)
async def _get_doctype_permissions(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.client.get_perm",
doctype=args["doctype"],
)
return json.dumps(result, indent=2)
async def _list_modules(args: dict) -> str:
client = FrappeClient()
result = await client.get_list(
"Module Def",
fields=["name", "app_name"],
limit=150,
order_by="name asc",
)
return json.dumps(result, indent=2)
async def _get_doctype_meta(args: dict) -> str:
client = FrappeClient()
meta = await client.call_method(
"frappe.desk.form.load.getdoctype",
doctype=args["doctype"],
with_parent=1,
cached_timestamp=None,
)
if args.get("include_permissions", True):
try:
perms = await client.call_method("frappe.client.get_perm", doctype=args["doctype"])
if isinstance(meta, dict):
meta["_permissions"] = perms
except Exception:
pass
return json.dumps(meta, indent=2)

View File

@ -0,0 +1,425 @@
"""
Level 10 Safety and governance tools.
Dry run, validate, risk scoring, confirmation tokens, audit log, rollback tracking.
"""
import json
import time
import hashlib
import secrets
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
from frappe_mcp.audit_store import log_action, read_audit_log, clear_audit_log
# In-memory confirmation token store (clears on restart — by design)
_pending_confirmations: dict[str, dict] = {}
# Reversible action registry: tool_call_id → undo info
_reversible_actions: dict[str, dict] = {}
# Risk profiles per tool prefix
_RISK_MAP = {
"frappe_delete": "high",
"frappe_bulk_delete": "high",
"frappe_cancel": "high",
"frappe_reset_customization": "high",
"frappe_revoke_api_key": "high",
"frappe_clear_error_logs": "medium",
"frappe_bulk_submit": "medium",
"frappe_submit": "medium",
"frappe_create": "low",
"frappe_update": "low",
"frappe_get": "low",
"frappe_list": "low",
"frappe_ping": "low",
}
def _score_risk(tool_name: str, arguments: dict) -> str:
for prefix, level in _RISK_MAP.items():
if tool_name.startswith(prefix):
return level
if "delete" in tool_name or "cancel" in tool_name or "reset" in tool_name:
return "high"
if "submit" in tool_name or "bulk" in tool_name:
return "medium"
return "low"
def tools() -> list[Tool]:
return [
Tool(
name="frappe_validate_document_payload",
description=(
"Validate document fields before creating or updating. "
"Checks required fields, field types, and link validity. "
"Does NOT save anything."
),
inputSchema={
"type": "object",
"required": ["doctype", "data"],
"properties": {
"doctype": {"type": "string"},
"data": {"type": "object"},
"name": {"type": "string", "description": "Provide if validating an update"},
},
},
),
Tool(
name="frappe_dry_run_action",
description=(
"Simulate a create or update action without saving. "
"Returns what WOULD happen: validation result, computed fields, and any errors."
),
inputSchema={
"type": "object",
"required": ["action", "doctype", "data"],
"properties": {
"action": {"type": "string", "enum": ["create", "update"]},
"doctype": {"type": "string"},
"data": {"type": "object"},
"name": {"type": "string", "description": "Required for update dry-run"},
},
},
),
Tool(
name="frappe_risk_score_action",
description=(
"Score the risk level of a proposed action before executing it. "
"Returns low | medium | high with reasoning."
),
inputSchema={
"type": "object",
"required": ["tool_name", "arguments"],
"properties": {
"tool_name": {"type": "string"},
"arguments": {"type": "object"},
},
},
),
Tool(
name="frappe_request_confirmation_token",
description=(
"Generate a one-time confirmation token for a high-risk action. "
"The token expires in 60 seconds. "
"Pass the token to frappe_confirm_and_execute to proceed."
),
inputSchema={
"type": "object",
"required": ["tool_name", "arguments"],
"properties": {
"tool_name": {"type": "string"},
"arguments": {"type": "object"},
"reason": {"type": "string", "description": "Why this action is being performed"},
},
},
),
Tool(
name="frappe_audit_tool_call",
description="Manually log a tool call to the MCP audit log.",
inputSchema={
"type": "object",
"required": ["tool_name", "arguments"],
"properties": {
"tool_name": {"type": "string"},
"arguments": {"type": "object"},
"result_summary": {"type": "string"},
"risk": {"type": "string", "enum": ["low", "medium", "high"], "default": "low"},
},
},
),
Tool(
name="frappe_get_audit_history",
description="Read the MCP-level audit log of all tool calls made in this session.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "default": 50},
"tool_filter": {"type": "string", "description": "Filter by tool name substring"},
},
},
),
Tool(
name="frappe_clear_audit_log",
description="Clear the local MCP audit log file.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_tool_health_check",
description=(
"Full health check: Frappe connectivity, auth, read/write access, "
"scheduler status, and MCP module status."
),
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="frappe_policy_check",
description=(
"Check whether an action is allowed under the current MCP policy "
"(read_only_mode, enabled modules, risk thresholds)."
),
inputSchema={
"type": "object",
"required": ["tool_name"],
"properties": {
"tool_name": {"type": "string"},
"arguments": {"type": "object"},
},
},
),
Tool(
name="frappe_register_reversible_action",
description=(
"Register an action as reversible so it can be rolled back later. "
"Stores doctype, name, and the pre-action state."
),
inputSchema={
"type": "object",
"required": ["action_id", "doctype", "name"],
"properties": {
"action_id": {"type": "string"},
"doctype": {"type": "string"},
"name": {"type": "string"},
"pre_state": {"type": "object", "description": "Document state before the action"},
},
},
),
Tool(
name="frappe_rollback_action",
description=(
"Roll back a previously registered reversible action "
"by restoring the document to its pre-action state."
),
inputSchema={
"type": "object",
"required": ["action_id"],
"properties": {
"action_id": {"type": "string"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_validate_document_payload": _validate_payload,
"frappe_dry_run_action": _dry_run_action,
"frappe_risk_score_action": _risk_score_action,
"frappe_request_confirmation_token": _request_confirmation_token,
"frappe_audit_tool_call": _audit_tool_call,
"frappe_get_audit_history": _get_audit_history,
"frappe_clear_audit_log": _clear_audit_log,
"frappe_tool_health_check": _tool_health_check,
"frappe_policy_check": _policy_check,
"frappe_register_reversible_action": _register_reversible_action,
"frappe_rollback_action": _rollback_action,
}
async def _validate_payload(args: dict) -> str:
client = FrappeClient()
meta = await client.call_method(
"frappe.desk.form.load.getdoctype",
doctype=args["doctype"],
with_parent=1,
cached_timestamp=None,
)
data = args["data"]
errors = []
warnings = []
docs = meta.get("docs", []) if isinstance(meta, dict) else []
fields = next((d.get("fields", []) for d in docs if d.get("name") == args["doctype"]), [])
for field in fields:
fname = field.get("fieldname", "")
if field.get("reqd") and fname not in data and not data.get(fname):
errors.append(f"Required field missing: {fname} ({field.get('label', fname)})")
if fname in data and field.get("fieldtype") == "Int":
try:
int(data[fname])
except (TypeError, ValueError):
errors.append(f"Field {fname} expects an integer, got: {data[fname]}")
return json.dumps({
"valid": len(errors) == 0,
"errors": errors,
"warnings": warnings,
"fields_provided": list(data.keys()),
}, indent=2)
async def _dry_run_action(args: dict) -> str:
client = FrappeClient()
action = args["action"]
doctype = args["doctype"]
data = args["data"]
try:
if action == "create":
result = await client.call_method(
"frappe.client.validate",
doc={"doctype": doctype, **data},
)
else:
name = args.get("name", "")
existing = await client.get_doc(doctype, name)
merged = {**existing, **data}
result = await client.call_method("frappe.client.validate", doc=merged)
return json.dumps({"dry_run": True, "action": action, "status": "would_succeed", "result": result}, indent=2)
except Exception as e:
return json.dumps({"dry_run": True, "action": action, "status": "would_fail", "error": str(e)}, indent=2)
async def _risk_score_action(args: dict) -> str:
tool = args["tool_name"]
arguments = args.get("arguments", {})
risk = _score_risk(tool, arguments)
reasons = []
if "delete" in tool:
reasons.append("Destructive operation — cannot be undone easily")
if "bulk" in tool:
reasons.append("Bulk operation affects multiple records")
if "cancel" in tool:
reasons.append("Cancellation changes document lifecycle")
if "reset" in tool:
reasons.append("Resets customization — removes all property setters and custom fields")
if not reasons:
reasons.append("Standard read/write operation")
return json.dumps({"tool": tool, "risk": risk, "reasons": reasons}, indent=2)
async def _request_confirmation_token(args: dict) -> str:
token = secrets.token_hex(8)
_pending_confirmations[token] = {
"tool_name": args["tool_name"],
"arguments": args["arguments"],
"reason": args.get("reason", ""),
"expires_at": time.time() + 60,
"created_at": time.time(),
}
return json.dumps({
"token": token,
"tool_name": args["tool_name"],
"expires_in_seconds": 60,
"instruction": "Pass this token to frappe_confirm_and_execute to proceed with the action.",
}, indent=2)
async def _audit_tool_call(args: dict) -> str:
record_id = log_action(
tool_name=args["tool_name"],
arguments=args["arguments"],
result_summary=args.get("result_summary", ""),
risk=args.get("risk", "low"),
)
return json.dumps({"logged": True, "record_id": record_id}, indent=2)
async def _get_audit_history(args: dict) -> str:
records = read_audit_log(
limit=args.get("limit", 50),
tool_filter=args.get("tool_filter", ""),
)
return json.dumps({"count": len(records), "records": records}, indent=2)
async def _clear_audit_log(args: dict) -> str:
count = clear_audit_log()
return json.dumps({"cleared": True, "records_removed": count}, indent=2)
async def _tool_health_check(args: dict) -> str:
from frappe_mcp.healthcheck import run_health_check
from frappe_mcp.module_registry import ALL_MODULES, _is_module_enabled
from frappe_mcp.config import get_settings
import asyncio
client = FrappeClient()
settings = get_settings()
checks = {}
# connectivity
try:
versions = await client.call_method("frappe.utils.change_log.get_versions")
checks["frappe_reachable"] = {"status": "ok", "version": versions.get("frappe", {}).get("version") if isinstance(versions, dict) else None}
except Exception as e:
checks["frappe_reachable"] = {"status": "fail", "error": str(e)}
# auth
try:
user = await client.call_method("frappe.auth.get_logged_user")
checks["auth"] = {"status": "ok", "user": user}
except Exception as e:
checks["auth"] = {"status": "fail", "error": str(e)}
# modules
checks["modules"] = {
key: "enabled" if _is_module_enabled(key) else "disabled"
for key, _, _ in ALL_MODULES
}
checks["read_only_mode"] = settings.read_only_mode
checks["total_enabled_tools"] = sum(
len(mod.tools()) for key, mod, _ in ALL_MODULES if _is_module_enabled(key)
)
return json.dumps(checks, indent=2)
async def _policy_check(args: dict) -> str:
from frappe_mcp.config import get_settings
from frappe_mcp.module_registry import _is_module_enabled, ALL_MODULES
settings = get_settings()
tool = args["tool_name"]
risk = _score_risk(tool, args.get("arguments", {}))
violations = []
if settings.read_only_mode:
write_prefixes = ["frappe_create", "frappe_update", "frappe_delete", "frappe_submit",
"frappe_cancel", "frappe_bulk", "erpnext_"]
if any(tool.startswith(p) for p in write_prefixes):
violations.append("read_only_mode=true — write operations are blocked")
module_key = None
for key, mod, _ in ALL_MODULES:
if tool in mod.handlers():
module_key = key
break
if module_key and not _is_module_enabled(module_key):
violations.append(f"Module '{module_key}' is disabled in configuration")
return json.dumps({
"tool": tool,
"risk": risk,
"allowed": len(violations) == 0,
"violations": violations,
}, indent=2)
async def _register_reversible_action(args: dict) -> str:
_reversible_actions[args["action_id"]] = {
"doctype": args["doctype"],
"name": args["name"],
"pre_state": args.get("pre_state", {}),
"registered_at": time.time(),
}
return json.dumps({"registered": True, "action_id": args["action_id"]}, indent=2)
async def _rollback_action(args: dict) -> str:
action = _reversible_actions.get(args["action_id"])
if not action:
return json.dumps({"error": f"No reversible action found for id: {args['action_id']}"}, indent=2)
client = FrappeClient()
try:
result = await client.update_doc(action["doctype"], action["name"], action["pre_state"])
del _reversible_actions[args["action_id"]]
return json.dumps({"rolled_back": True, "doctype": action["doctype"], "name": action["name"]}, indent=2)
except Exception as e:
return json.dumps({"rolled_back": False, "error": str(e)}, indent=2)

View File

@ -0,0 +1,111 @@
"""Naming Series tools — manage document naming patterns in Frappe."""
import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
def tools() -> list[Tool]:
return [
Tool(
name="frappe_get_naming_series",
description="Get all naming series options for a DocType.",
inputSchema={
"type": "object",
"required": ["doctype"],
"properties": {
"doctype": {"type": "string"},
},
},
),
Tool(
name="frappe_set_naming_series",
description=(
"Set or update naming series options for a DocType. "
"Series format: 'PREFIX-.YYYY.-.MM.-.####' where #### is zero-padded counter."
),
inputSchema={
"type": "object",
"required": ["doctype", "series"],
"properties": {
"doctype": {"type": "string"},
"series": {
"type": "string",
"description": "Newline-separated list of series patterns e.g. 'INV-.YYYY.-.####\\nINV-TEST-.####'",
},
},
},
),
Tool(
name="frappe_get_series_counter",
description="Get the current counter value for a naming series prefix.",
inputSchema={
"type": "object",
"required": ["prefix"],
"properties": {
"prefix": {"type": "string", "description": "Prefix e.g. 'INV-2024-'"},
},
},
),
Tool(
name="frappe_update_series_counter",
description="Manually set the counter for a naming series prefix.",
inputSchema={
"type": "object",
"required": ["prefix", "value"],
"properties": {
"prefix": {"type": "string"},
"value": {"type": "integer", "description": "New counter value"},
},
},
),
]
def handlers() -> dict:
return {
"frappe_get_naming_series": _get_naming_series,
"frappe_set_naming_series": _set_naming_series,
"frappe_get_series_counter": _get_series_counter,
"frappe_update_series_counter": _update_series_counter,
}
async def _get_naming_series(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.core.doctype.naming_series.naming_series.get_options",
arg=args["doctype"],
)
return json.dumps(result, indent=2)
async def _set_naming_series(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.core.doctype.naming_series.naming_series.update_series",
args={
"df": {"options": args["series"]},
"doctype": args["doctype"],
},
)
return json.dumps(result, indent=2)
async def _get_series_counter(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.core.doctype.naming_series.naming_series.get_current",
prefix=args["prefix"],
)
return json.dumps({"prefix": args["prefix"], "current": result}, indent=2)
async def _update_series_counter(args: dict) -> str:
client = FrappeClient()
result = await client.call_method(
"frappe.core.doctype.naming_series.naming_series.update_counter",
prefix=args["prefix"],
value=args["value"],
)
return json.dumps(result, indent=2)

Some files were not shown because too many files have changed in this diff Show More