create_neo4j_destination
Establish a Neo4j destination connector by specifying database details, URI, and credentials for integration with the Unstructured API MCP Server.
Instructions
Create an neo4j destination connector.
Args:
name: A unique name for this connector
database: The neo4j database, e.g. "neo4j"
uri: The neo4j URI, e.g. neo4j+s://<neo4j_instance_id>.databases.neo4j.io
username: The neo4j username
Returns:
String containing the created destination connector information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| batch_size | No | ||
| database | Yes | ||
| name | Yes | ||
| uri | Yes | ||
| username | Yes |
Implementation Reference
- The handler function that executes the tool logic: prepares Neo4j config, creates the destination connector via the client API, and logs the result.async def create_neo4j_destination( ctx: Context, name: str, database: str, uri: str, batch_size: Optional[int] = 100, ) -> str: """Create an neo4j destination connector. Args: name: A unique name for this connector database: The neo4j database, e.g. "neo4j" uri: The neo4j URI, e.g. neo4j+s://<neo4j_instance_id>.databases.neo4j.io batch_size: The batch size for the connector Returns: String containing the created destination connector information """ client = ctx.request_context.lifespan_context.client config = _prepare_neo4j_dest_config(database, uri, batch_size) destination_connector = CreateDestinationConnector(name=name, type="neo4j", 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="neo4j", connector_type="Destination", created_or_updated="Created", ) return result except Exception as e: return f"Error creating neo4j destination connector: {str(e)}"
- Supporting utility to construct the Neo4jDestinationConnectorConfigInput from parameters and environment variables.def _prepare_neo4j_dest_config( database: str, uri: str, batch_size: Optional[int] = None, ) -> Neo4jDestinationConnectorConfigInput: """Prepare the Azure source connector configuration.""" if os.getenv("NEO4J_USERNAME") is None: raise ValueError("NEO4J_USERNAME environment variable is not set") if os.getenv("NEO4J_PASSWORD") is None: raise ValueError("NEO4J_PASSWORD environment variable is not set") else: return Neo4jDestinationConnectorConfigInput( database=database, uri=uri, batch_size=batch_size, username=os.getenv("NEO4J_USERNAME"), password=os.getenv("NEO4J_PASSWORD"), )
- uns_mcp/connectors/destination/destination_tool.py:109-127 (registration)Dispatches to create_neo4j_destination when destination_type is 'neo4j' in the create_destination_connector tool. This is the entry point where the specific handler is invoked.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, } if destination_type in destination_functions: destination_function = destination_functions[destination_type] return await destination_function(ctx=ctx, name=name, **type_specific_config) return ( f"Unsupported destination type: {destination_type}. " f"Please use a supported destination type {list(destination_functions.keys())}." )
- uns_mcp/connectors/destination/destination_tool.py:23-26 (registration)Imports the create_neo4j_destination handler for use in the dispatcher.from uns_mcp.connectors.destination.neo4j import ( create_neo4j_destination, update_neo4j_destination, )