create_s3_destination
Create an S3 destination connector for storing data by specifying S3 bucket details, AWS credentials, and optional session tokens or custom endpoints.
Instructions
Create an S3 destination connector.
Args:
name: A unique name for this connector
remote_url: The S3 URI to the bucket or folder
key: The AWS access key ID
secret: The AWS secret access key
token: The AWS STS session token for temporary access (optional)
endpoint_url: Custom URL if connecting to a non-AWS S3 bucket
Returns:
String containing the created destination connector information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| remote_url | Yes |
Implementation Reference
- The handler function that implements the logic to create an S3 destination connector using the unstructured_client API.async def create_s3_destination( ctx: Context, name: str, remote_url: str, ) -> str: """Create an S3 destination connector. Args: name: A unique name for this connector remote_url: The S3 URI to the bucket or folder Returns: String containing the created destination connector information """ client = ctx.request_context.lifespan_context.client config = _prepare_s3_dest_config(remote_url) destination_connector = CreateDestinationConnector( name=name, type=DestinationConnectorType.S3, config=config, ) try: response = await client.destinations.create_destination_async( request=CreateDestinationRequest(create_destination_connector=destination_connector), ) result = create_log_for_created_updated_connector( response, connector_name="S3", connector_type="Destination", created_or_updated="Created", ) return result except Exception as e: return f"Error creating S3 destination connector: {str(e)}"
- Helper function to prepare the S3 destination configuration from environment variables.def _prepare_s3_dest_config( remote_url: Optional[str], ) -> S3DestinationConnectorConfigInput: """Prepare the S3 destination connector configuration.""" config = S3DestinationConnectorConfigInput( remote_url=remote_url, key=os.getenv("AWS_KEY"), secret=os.getenv("AWS_SECRET"), ) if os.getenv("TOKEN"): config.token = os.getenv("TOKEN") if os.getenv("ENDPOINT_URL"): config.endpoint_url = os.getenv("ENDPOINT_URL") return config
- uns_mcp/connectors/destination/destination_tool.py:109-118 (registration)Registration of create_s3_destination in the dispatch dictionary of the create_destination_connector tool for the 's3' destination type.destination_functions = { "astradb": create_astradb_destination, "databricks_delta_table": create_databricks_delta_table_destination, "databricks_volumes": create_databricks_volumes_destination, "mongodb": create_mongodb_destination, "neo4j": create_neo4j_destination, "pinecone": create_pinecone_destination, "s3": create_s3_destination, "weaviate": create_weaviate_destination, }
- uns_mcp/connectors/destination/destination_tool.py:31-34 (registration)Import of the create_s3_destination handler function.from uns_mcp.connectors.destination.s3 import ( create_s3_destination, update_s3_destination, )
- uns_mcp/connectors/destination/__init__.py:10-14 (registration)MCP tool registration function that registers the dispatcher tools including access to create_s3_destination.def register_destination_connectors(mcp: FastMCP): """Register all destination connector tools with the MCP server.""" mcp.tool()(create_destination_connector) mcp.tool()(update_destination_connector) mcp.tool()(delete_destination_connector)