# MCP Operations - Area-Based Architecture
This directory contains the area-based MCP operations for Microsoft Graph API.
## Structure
```
operations/
├── __init__.py # Module exports
├── base.py # BaseOperation abstract class
├── user_operations.py # User management (5 actions)
├── calendar_operations.py # Calendar & events (7 actions)
├── email_operations.py # Email management (10 actions)
├── teams_operations.py # Teams & chats (7 actions)
└── file_operations.py # Files & documents (8 actions)
```
## Base Operation Class
All operations inherit from `BaseOperation` which provides:
- **Unified execution**: `execute(action, params)` method
- **Validation**: Parameter validation before API calls
- **Error handling**: Standardized error responses with clear codes
- **Response formatting**: Consistent response structure across all operations
### Example Implementation
```python
from .base import BaseOperation, OperationError
class MyOperations(BaseOperation):
def get_supported_actions(self):
return ["action1", "action2", "action3"]
def _validate_action_params(self, action, params):
if action == "action1":
self._require_param(params, "required_field", str)
# Add more validation...
def _execute_action(self, action, params):
if action == "action1":
return self._action_action1(params)
# Handle other actions...
def _action_action1(self, params):
# Implement action logic
result = self.graph_client.get("/some/endpoint")
return {"result": result}
```
## Validation Helpers
The base class provides validation helpers:
- `_require_param(params, name, type)` - Check required parameter with optional type check
- `_validate_email(email)` - Validate email address format
- `_validate_datetime(dt_str, param_name)` - Validate ISO 8601 datetime
- `_validate_choice(value, choices, param_name)` - Validate value is in allowed choices
## Error Codes
Standard error codes across all operations:
### Generic Errors
- `INVALID_ACTION` - Action not supported
- `MISSING_PARAM` - Required parameter missing
- `INVALID_PARAM_TYPE` - Parameter has wrong type
- `INVALID_PARAM` - Parameter value invalid
- `INTERNAL_ERROR` - Unexpected error
### Validation Errors
- `INVALID_EMAIL` - Email address format invalid
- `INVALID_DATETIME` - Datetime format invalid
- `INVALID_CHOICE` - Value not in allowed choices
- `INVALID_QUERY` - Search query too short or invalid
- `INVALID_TIME_RANGE` - End time before start time
### Domain-Specific Errors
- `USER_NOT_FOUND` - User doesn't exist
- `EVENT_NOT_FOUND` - Calendar event doesn't exist
- `MESSAGE_NOT_FOUND` - Email message doesn't exist
- `CHAT_NOT_FOUND` - Teams chat doesn't exist
- `FILE_NOT_FOUND` - File doesn't exist
- `NO_SLOTS_FOUND` - No common availability slots found
- `SEND_FAILED` - Failed to send email
- `CREATE_FAILED` - Failed to create resource
- `UPDATE_FAILED` - Failed to update resource
- `DELETE_FAILED` - Failed to delete resource
## Response Format
All operations return a standardized response:
### Success Response
```json
{
"success": true,
"data": {
// Operation-specific data
},
"metadata": {
"action": "search",
"area": "user_operations",
"timestamp": "2025-10-15T08:45:00Z"
}
}
```
### Error Response
```json
{
"success": false,
"error": {
"code": "MISSING_PARAM",
"message": "Required parameter 'subject' not provided",
"details": {
"param": "subject"
}
},
"metadata": {
"action": "create",
"area": "calendar_operations",
"timestamp": "2025-10-15T08:45:00Z"
}
}
```
## Adding New Actions
To add a new action to an existing operation:
1. **Add action name** to `get_supported_actions()`:
```python
def get_supported_actions(self):
return ["existing1", "existing2", "new_action"]
```
2. **Add validation** in `_validate_action_params()`:
```python
def _validate_action_params(self, action, params):
# ... existing validations
elif action == "new_action":
self._require_param(params, "field1", str)
# Add more validation...
```
3. **Add execution handler** in `_execute_action()`:
```python
def _execute_action(self, action, params):
# ... existing handlers
elif action == "new_action":
return self._action_new_action(params)
```
4. **Implement action logic**:
```python
def _action_new_action(self, params):
# Implement your logic
result = self.graph_client.get("/endpoint")
return {"data": result}
```
That's it! The action is automatically:
- Available via the API endpoint
- Listed in `/mcp/info` endpoint
- Included in validation
- Using standardized error handling
- Formatted with consistent response structure
## Usage in Server
Operations are instantiated in `server_v2.py`:
```python
from .operations import UserOperations, CalendarOperations, ...
graph_client = MicrosoftGraphClient()
user_ops = UserOperations(graph_client)
calendar_ops = CalendarOperations(graph_client)
@app.post("/mcp/user_operations")
async def user_operations(request: OperationRequest):
result = user_ops.execute(request.action, request.params)
if not result["success"]:
raise HTTPException(status_code=400, detail=result)
return result
```
## Testing
Test operations using the test suite:
```bash
python test_area_based_mcp.py
```
Or test individual operations via curl:
```bash
# Search users
curl -X POST http://localhost:8001/mcp/user_operations \
-H "Content-Type: application/json" \
-d '{"action": "search", "params": {"query": "Giovanni", "max_results": 5}}'
# Create calendar event
curl -X POST http://localhost:8001/mcp/calendar_operations \
-H "Content-Type: application/json" \
-d '{
"action": "create",
"params": {
"subject": "Team Meeting",
"start": "2025-10-20T10:00:00Z",
"end": "2025-10-20T11:00:00Z",
"is_online": true
}
}'
```
## Documentation
Full architecture documentation: `docs/MCP_AREA_BASED_ARCHITECTURE.md`
Interactive API docs when server running:
- Swagger UI: http://localhost:8001/docs
- ReDoc: http://localhost:8001/redoc
- Info endpoint: http://localhost:8001/mcp/info
