Unstructured API MCP Server

Instructions

Implementation Reference

• uns_mcp/connectors/source/source_tool.py:89-155 (handler)

  The core handler function for the 'update_source_connector' MCP tool. It takes source_id, source_type, and type_specific_config, dispatches to the appropriate type-specific update function (imported from individual connector modules), and returns a status message. Includes detailed docstring serving as input schema documentation.
  async def update_source_connector( ctx: Context, source_id: str, source_type: Literal["azure", "onedrive", "salesforce", "gdrive", "s3", "sharepoint"], type_specific_config: dict[str, Any], ) -> str: """Update a source connector based on type. Args: ctx: Context object with the request and lifespan context source_id: ID of the source connector to update source_type: The type of source being updated (e.g., 'azure', 'onedrive', 'salesforce', 'gdrive', 's3', 'sharepoint') type_specific_config: azure: remote_url: (Optional[str]) The Azure Storage remote URL with the format az://<container-name>/<path/to/file/or/folder/in/container/as/needed> recursive: (Optional[bool]) Whether to access subfolders gdrive: drive_id: (Optional[str]) The Drive ID for the Google Drive source recursive: (Optional[bool]) Whether to access subfolders extensions: (Optional[list[str]]) File extensions to filter onedrive: path: (Optional[str]) The path to the target folder in the OneDrive account user_pname: (Optional[str]) The User Principal Name (UPN) for the OneDrive user account recursive: (Optional[bool]) Whether to access subfolders authority_url: (Optional[str]) The authentication token provider URL s3: remote_url: (Optional[str]) The S3 URI to the bucket or folder (e.g., s3://my-bucket/) recursive: (Optional[bool]) Whether to access subfolders salesforce: username: (Optional[str]) The Salesforce username categories: (Optional[list[str]]) Optional Salesforce domain,the names of the Salesforce categories (objects) that you want to access, specified as a comma-separated list. Available categories include Account, Campaign, Case, EmailMessage, and Lead. sharepoint: site: Optional([str]) The SharePoint site to connect to user_pname: Optional([str]) The username for the SharePoint site path: (Optional) The path within the SharePoint site recursive: (Optional[bool]) Whether to access subfolders authority_url: (Optional[str]) The authority URL for authentication Returns: String containing the updated source connector information """ update_functions = { "azure": update_azure_source, "gdrive": update_gdrive_source, "onedrive": update_onedrive_source, "s3": update_s3_source, "salesforce": update_salesforce_source, "sharepoint": update_sharepoint_source, } if source_type in update_functions: update_function = update_functions[source_type] return await update_function(ctx=ctx, source_id=source_id, **type_specific_config) return ( f"Unsupported source type: {source_type}. " f"Please use a supported source type: {list(update_functions.keys())}." )
• uns_mcp/connectors/source/__init__.py:13-15 (registration)

  Registration of the 'update_source_connector' tool (along with related create/delete tools) using FastMCP's mcp.tool() decorator within the register_source_connectors function.
  mcp.tool()(create_source_connector) mcp.tool()(update_source_connector) mcp.tool()(delete_source_connector)
• uns_mcp/connectors/source/source_tool.py:89-155 (schema)

  Type hints and comprehensive docstring in the handler function define the input schema, including parameters and type-specific configuration options for each supported source_type.
  async def update_source_connector( ctx: Context, source_id: str, source_type: Literal["azure", "onedrive", "salesforce", "gdrive", "s3", "sharepoint"], type_specific_config: dict[str, Any], ) -> str: """Update a source connector based on type. Args: ctx: Context object with the request and lifespan context source_id: ID of the source connector to update source_type: The type of source being updated (e.g., 'azure', 'onedrive', 'salesforce', 'gdrive', 's3', 'sharepoint') type_specific_config: azure: remote_url: (Optional[str]) The Azure Storage remote URL with the format az://<container-name>/<path/to/file/or/folder/in/container/as/needed> recursive: (Optional[bool]) Whether to access subfolders gdrive: drive_id: (Optional[str]) The Drive ID for the Google Drive source recursive: (Optional[bool]) Whether to access subfolders extensions: (Optional[list[str]]) File extensions to filter onedrive: path: (Optional[str]) The path to the target folder in the OneDrive account user_pname: (Optional[str]) The User Principal Name (UPN) for the OneDrive user account recursive: (Optional[bool]) Whether to access subfolders authority_url: (Optional[str]) The authentication token provider URL s3: remote_url: (Optional[str]) The S3 URI to the bucket or folder (e.g., s3://my-bucket/) recursive: (Optional[bool]) Whether to access subfolders salesforce: username: (Optional[str]) The Salesforce username categories: (Optional[list[str]]) Optional Salesforce domain,the names of the Salesforce categories (objects) that you want to access, specified as a comma-separated list. Available categories include Account, Campaign, Case, EmailMessage, and Lead. sharepoint: site: Optional([str]) The SharePoint site to connect to user_pname: Optional([str]) The username for the SharePoint site path: (Optional) The path within the SharePoint site recursive: (Optional[bool]) Whether to access subfolders authority_url: (Optional[str]) The authority URL for authentication Returns: String containing the updated source connector information """ update_functions = { "azure": update_azure_source, "gdrive": update_gdrive_source, "onedrive": update_onedrive_source, "s3": update_s3_source, "salesforce": update_salesforce_source, "sharepoint": update_sharepoint_source, } if source_type in update_functions: update_function = update_functions[source_type] return await update_function(ctx=ctx, source_id=source_id, **type_specific_config) return ( f"Unsupported source type: {source_type}. " f"Please use a supported source type: {list(update_functions.keys())}." )