create_weaviate_destination
Set up a Weaviate vector database destination connector by specifying the cluster URL and collection for efficient data management and retrieval.
Instructions
Create an weaviate vector database destination connector.
Args:
cluster_url: URL of the weaviate cluster
collection : Name of the collection to use in the weaviate cluster
Note: The collection is a table in the weaviate cluster.
In platform, there are dedicated code to generate collection for users
here, due to the simplicity of the server, we are not generating it for users.
Returns:
String containing the created destination connector information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| cluster_url | Yes | ||
| collection | Yes | ||
| name | Yes |
Implementation Reference
- The main asynchronous handler function that creates a Weaviate destination connector by preparing config, calling the client API, and logging the result.async def create_weaviate_destination( ctx: Context, name: str, cluster_url: str, collection: str, ) -> str: """Create an weaviate vector database destination connector. Args: cluster_url: URL of the weaviate cluster collection : Name of the collection to use in the weaviate cluster Note: The collection is a table in the Weaviate cluster. In the platform, the collection name can be generated automatically; however, this is not yet supported via the API. Therefore, the collection name is required. Please note that the schema cannot be empty and must contain at least a record_id: Text. Returns: String containing the created destination connector information """ client = ctx.request_context.lifespan_context.client config = _prepare_weaviate_dest_config(collection, cluster_url) destination_connector = CreateDestinationConnector( name=name, type=DestinationConnectorType.WEAVIATE_CLOUD, 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="Weaviate", connector_type="Destination", created_or_updated="Created", ) return result except Exception as e: return f"Error creating weaviate destination connector: {str(e)}"
- Helper function to prepare and return the WeaviateDestinationConnectorConfigInput object using provided parameters and environment variable for API key.def _prepare_weaviate_dest_config( collection: str, cluster_url: str, ) -> WeaviateDestinationConnectorConfigInput: """Prepare the Azure source connector configuration.""" return WeaviateDestinationConnectorConfigInput( cluster_url=cluster_url, api_key=os.getenv("WEAVIATE_CLOUD_API_KEY"), collection=collection, )
- uns_mcp/connectors/destination/destination_tool.py:109-118 (registration)Registration of create_weaviate_destination in the destination_functions dictionary, which dispatches to specific handlers based on destination_type in the create_destination_connector tool.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, }
- Schema definition including 'weaviate' as a valid destination_type in the Literal type for create_destination_connector tool.destination_type: Literal[ "astradb", "databricks_delta_table", "databricks_volumes", "mongodb", "neo4j", "pinecone", "s3", "weaviate", ],
- uns_mcp/connectors/destination/destination_tool.py:35-38 (registration)Import statement registering the create_weaviate_destination function for use in destination_tool.py.from uns_mcp.connectors.destination.weaviate import ( create_weaviate_destination, update_weaviate_destination, )