273 lines
9.1 KiB
Python
273 lines
9.1 KiB
Python
"""
|
|
Data access functions for Microsoft and Google services.
|
|
Provides standardized interfaces for SharePoint, Outlook, and other services.
|
|
"""
|
|
|
|
from typing import List, Dict, Any, Optional, Union
|
|
from datetime import datetime
|
|
from pydantic import BaseModel, Field
|
|
from enum import Enum
|
|
|
|
class ServiceType(str, Enum):
|
|
"""Service types for data access"""
|
|
MSFT = "msft"
|
|
GOOGLE = "google"
|
|
|
|
class FileRef(BaseModel):
|
|
"""Reference to a file in storage"""
|
|
id: str
|
|
name: str
|
|
path: str
|
|
url: Optional[str] = None
|
|
size: Optional[int] = None
|
|
lastModified: Optional[datetime] = None
|
|
|
|
# SharePoint Functions
|
|
class SharePointSearchParams(BaseModel):
|
|
"""Parameters for SharePoint search"""
|
|
userName: str
|
|
query: str
|
|
site: Optional[str] = None
|
|
folder: Optional[str] = None
|
|
contentType: Optional[str] = None
|
|
createdAfter: Optional[datetime] = None
|
|
modifiedAfter: Optional[datetime] = None
|
|
maxResults: Optional[int] = 100
|
|
|
|
class SharePointFolderParams(BaseModel):
|
|
"""Parameters for SharePoint folder operations"""
|
|
userName: str
|
|
folderPattern: str
|
|
site: Optional[str] = None
|
|
recursive: bool = False
|
|
includeFiles: bool = True
|
|
|
|
class SharePointFileParams(BaseModel):
|
|
"""Parameters for SharePoint file operations"""
|
|
userName: str
|
|
fileName: str
|
|
site: Optional[str] = None
|
|
folder: Optional[str] = None
|
|
content: Optional[bytes] = None
|
|
contentType: Optional[str] = None
|
|
|
|
async def Msft_Sharepoint_Search(params: SharePointSearchParams) -> List[Dict[str, Any]]:
|
|
"""Search SharePoint for files and folders matching criteria"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Sharepoint_GetFolders(params: SharePointFolderParams) -> Dict[str, Any]:
|
|
"""Get SharePoint folders matching pattern"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Sharepoint_GetFiles(params: SharePointFileParams) -> Dict[str, Any]:
|
|
"""Get SharePoint files matching pattern"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Sharepoint_GetFile(params: SharePointFileParams) -> Dict[str, Any]:
|
|
"""Get specific SharePoint file"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Sharepoint_PutFile(params: SharePointFileParams) -> FileRef:
|
|
"""Upload file to SharePoint"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
# Outlook Mail Functions
|
|
class OutlookMailParams(BaseModel):
|
|
"""Parameters for Outlook mail operations"""
|
|
userName: str
|
|
folder: Optional[str] = None
|
|
messageId: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
body: Optional[str] = None
|
|
to: Optional[List[str]] = None
|
|
cc: Optional[List[str]] = None
|
|
bcc: Optional[List[str]] = None
|
|
attachments: Optional[List[FileRef]] = None
|
|
searchString: Optional[str] = None
|
|
fromAddress: Optional[str] = None
|
|
receivedAfter: Optional[datetime] = None
|
|
maxResults: Optional[int] = 100
|
|
|
|
async def Msft_Outlook_ReadMails(params: OutlookMailParams) -> List[Dict[str, Any]]:
|
|
"""Read multiple emails from Outlook"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_ReadMail(params: OutlookMailParams) -> Dict[str, Any]:
|
|
"""Read specific email from Outlook"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_DraftMail(params: OutlookMailParams) -> Dict[str, Any]:
|
|
"""Create draft email in Outlook"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_SendMail(params: OutlookMailParams) -> Dict[str, Any]:
|
|
"""Send email through Outlook"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
# Outlook Calendar Functions
|
|
class OutlookCalendarParams(BaseModel):
|
|
"""Parameters for Outlook calendar operations"""
|
|
userName: str
|
|
calendar: Optional[str] = None
|
|
eventId: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
body: Optional[str] = None
|
|
startTime: Optional[datetime] = None
|
|
endTime: Optional[datetime] = None
|
|
location: Optional[str] = None
|
|
organizer: Optional[str] = None
|
|
attendees: Optional[List[str]] = None
|
|
searchString: Optional[str] = None
|
|
maxResults: Optional[int] = 100
|
|
|
|
async def Msft_Outlook_ReadAppointments(params: OutlookCalendarParams) -> List[Dict[str, Any]]:
|
|
"""Read multiple calendar appointments"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_CreateAppointment(params: OutlookCalendarParams) -> Dict[str, Any]:
|
|
"""Create new calendar appointment"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_ReadAppointment(params: OutlookCalendarParams) -> Dict[str, Any]:
|
|
"""Read specific calendar appointment"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_UpdateAppointment(params: OutlookCalendarParams) -> Dict[str, Any]:
|
|
"""Update existing calendar appointment"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
async def Msft_Outlook_DeleteAppointment(params: OutlookCalendarParams) -> bool:
|
|
"""Delete calendar appointment"""
|
|
# Implementation would go here
|
|
pass
|
|
|
|
def get_data_access_functions() -> List[Dict[str, Any]]:
|
|
"""
|
|
Dynamically generates a comprehensive list of all available data access functions
|
|
with their parameters for use in agent prompts.
|
|
"""
|
|
import inspect
|
|
import sys
|
|
|
|
functions = []
|
|
current_module = sys.modules[__name__]
|
|
|
|
# Get all functions in the module
|
|
for name, obj in inspect.getmembers(current_module):
|
|
# Check if it's a function and starts with Msft_ or Google_
|
|
if inspect.isfunction(obj) and (name.startswith('Msft_') or name.startswith('Google_')):
|
|
# Get function signature
|
|
sig = inspect.signature(obj)
|
|
|
|
# Get return type annotation
|
|
return_type = obj.__annotations__.get('return', 'Any')
|
|
if hasattr(return_type, '__origin__'):
|
|
return_type = str(return_type)
|
|
|
|
# Get parameter model class
|
|
param_model = None
|
|
for param in sig.parameters.values():
|
|
if param.annotation.__module__ == __name__:
|
|
param_model = param.annotation
|
|
break
|
|
|
|
# Determine authority from function name
|
|
authority = ServiceType.MSFT if name.startswith('Msft_') else ServiceType.GOOGLE
|
|
|
|
# Create function entry
|
|
function_entry = {
|
|
"name": name,
|
|
"description": obj.__doc__ or "",
|
|
"parameters": param_model.schema() if param_model else {},
|
|
"return_type": str(return_type),
|
|
"authority": authority
|
|
}
|
|
|
|
functions.append(function_entry)
|
|
|
|
return functions
|
|
|
|
class DataAccess:
|
|
"""Manages data access functions for different services"""
|
|
|
|
def __init__(self):
|
|
"""Initialize the data access manager"""
|
|
self.functions = get_data_access_functions()
|
|
self._initialize_functions()
|
|
|
|
def _initialize_functions(self):
|
|
"""Initialize function groups and metadata"""
|
|
# Group functions by service type
|
|
self.msft_functions = {}
|
|
self.google_functions = {}
|
|
|
|
for func in self.functions:
|
|
func_name = func['name']
|
|
# Get the actual function object
|
|
func_obj = globals()[func_name]
|
|
|
|
if func['authority'] == ServiceType.MSFT:
|
|
self.msft_functions[func_name] = func_obj
|
|
else:
|
|
self.google_functions[func_name] = func_obj
|
|
|
|
@property
|
|
def msft(self) -> Dict[str, Any]:
|
|
"""Get Microsoft service functions and metadata"""
|
|
return {
|
|
'functions': self.msft_functions,
|
|
'metadata': {
|
|
'name': 'Microsoft Services',
|
|
'description': 'Microsoft Office 365 and SharePoint services',
|
|
'functions': [f for f in self.functions if f['authority'] == ServiceType.MSFT]
|
|
}
|
|
}
|
|
|
|
@property
|
|
def google(self) -> Dict[str, Any]:
|
|
"""Get Google service functions and metadata"""
|
|
return {
|
|
'functions': self.google_functions,
|
|
'metadata': {
|
|
'name': 'Google Services',
|
|
'description': 'Google Workspace services',
|
|
'functions': [f for f in self.functions if f['authority'] == ServiceType.GOOGLE]
|
|
}
|
|
}
|
|
|
|
@property
|
|
def utils(self) -> Dict[str, Any]:
|
|
"""Get utility functions for data access"""
|
|
return {
|
|
'getAvailableFunctions': lambda: self.functions,
|
|
'getFunctionInfo': lambda name: next((f for f in self.functions if f['name'] == name), None),
|
|
'getServiceFunctions': lambda service_type: [f for f in self.functions if f['authority'] == service_type]
|
|
}
|
|
|
|
def to_service_object(self) -> Dict[str, Any]:
|
|
"""Convert to service object format"""
|
|
return {
|
|
'msft': self.msft,
|
|
'google': self.google,
|
|
'utils': self.utils
|
|
}
|
|
|
|
def get_data_access() -> DataAccess:
|
|
"""Get a singleton instance of the data access manager"""
|
|
if not hasattr(get_data_access, '_instance'):
|
|
get_data_access._instance = DataAccess()
|
|
return get_data_access._instance
|
|
|