(feat) Add metadata tools for Couchbase MCP server #6
Conversation
teetangh
commented
Apr 16, 2025
•
teetangh
commented
- Implemented tools to retrieve cluster and bucket information.
- Added functionality to list Full-Text Search (FTS) and N1QL indexes.
- Enhanced error handling and logging for tool operations.
- Implemented tools to retrieve cluster and bucket information. - Added functionality to list Full-Text Search (FTS) and N1QL indexes. - Enhanced error handling and logging for tool operations.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hello @teetangh, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
Summary of Changes
This pull request introduces new MCP tools for retrieving metadata from a Couchbase cluster and bucket. It includes functionalities to fetch cluster information, bucket settings, and lists of FTS and N1QL indexes. The changes involve adding new functions in src/mcp_server.py to expose these tools and creating a new module src/meta.py containing the helper functions to interact with the Couchbase cluster and retrieve the required metadata. Error handling and logging have been incorporated into the tool operations.
Highlights
- New MCP Tools: Introduces four new MCP tools:
get_cluster_info,get_bucket_info,list_fts_indexes, andlist_n1ql_indexes. - Metadata Retrieval: Adds functionality to retrieve cluster diagnostics, bucket settings, and lists of FTS and N1QL indexes.
- Error Handling and Logging: Implements error handling and logging within the new MCP tools to provide better diagnostics.
- New Module: Creates a new module
src/meta.pyto encapsulate the logic for interacting with the Couchbase cluster and retrieving metadata.
Changelog
Click here to see the changelog
- src/mcp_server.py
- Added imports for typing hints
DictandList. - Imported helper functions from the new
meta.pymodule:_get_cluster_info,_get_bucket_info,_list_fts_indexes, and_list_n1ql_indexes(lines 13-18). - Implemented
get_cluster_infoMCP tool to retrieve cluster diagnostics (lines 98-109). - Implemented
get_bucket_infoMCP tool to retrieve bucket settings (lines 112-122). - Implemented
list_fts_indexesMCP tool to list all Full-Text Search (FTS) indexes (lines 125-135). - Implemented
list_n1ql_indexesMCP tool to list all N1QL (Query) indexes for the current bucket (lines 138-151). - Added sectioning comment
# --- Metadata Tools ---to delineate the new tools (line 96). - Added sectioning comment
# --- Existing Tools ---to delineate the existing tools (line 154).
- Added imports for typing hints
- src/meta.py
- Created a new module
src/meta.pyto contain helper functions for metadata retrieval. - Implemented
_get_cluster_infofunction to retrieve cluster diagnostics (lines 20-42). - Implemented
_get_bucket_infofunction to retrieve bucket settings (lines 45-72). - Implemented
_list_fts_indexesfunction to list all Full-Text Search (FTS) indexes (lines 75-99). - Implemented
_list_n1ql_indexesfunction to list all N1QL (Query) indexes for a specific bucket (lines 102-127). - Added logging to each of the helper functions to capture errors during metadata retrieval (lines 37-42, 67-72, 94-99, 122-127).
- Created a new module
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist is currently in preview and may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments to provide feedback.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
A cluster's secrets,
Indexes whisper, buckets gleam,
Metadata flows.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
teetangh
changed the title
There was a problem hiding this comment.
Code Review
This pull request introduces new tools for retrieving cluster and bucket information, listing FTS and N1QL indexes, and enhances error handling and logging. The changes look good overall, but there are a few areas that could be improved for clarity and consistency.
Summary of Findings
- Error Handling Consistency: The error handling in the
meta.pyfile is good, but it could be more consistent across all functions. Specifically, ensuring that all exceptions are logged with sufficient context and re-raised to allow the MCP framework to handle them. - Duplicated code: There is duplicated code in the new tool functions in
mcp_server.pythat could be refactored into a common helper function.
Merge Readiness
The pull request is almost ready for merging. Addressing the error handling consistency and potential code duplication would improve the overall quality and maintainability of the code. I am unable to directly approve the pull request, and recommend that others review and approve this code before merging.
| def get_cluster_info(ctx: Context) -> Dict[str, Any]: | ||
| """Get diagnostic information about the Couchbase cluster.""" | ||
| cluster = ctx.request_context.lifespan_context.cluster | ||
| if not cluster: | ||
| raise ValueError("Cluster connection not available in context.") | ||
| try: | ||
| return _get_cluster_info(cluster) | ||
| except Exception as e: | ||
| logger.error(f"Tool error getting cluster info: {type(e).__name__} - {e}") | ||
| # Re-raise to signal failure to the MCP framework/caller | ||
| raise |
There was a problem hiding this comment.
Consider extracting the cluster availability check and exception handling into a reusable helper function, as this pattern is repeated in multiple tool functions. This would reduce code duplication and improve maintainability. For example, you could have a function that takes a context and returns the cluster, raising an exception if it's not available, and then use that function in each tool.
| def get_cluster_info(ctx: Context) -> Dict[str, Any]: | |
| """Get diagnostic information about the Couchbase cluster.""" | |
| cluster = ctx.request_context.lifespan_context.cluster | |
| if not cluster: | |
| raise ValueError("Cluster connection not available in context.") | |
| try: | |
| return _get_cluster_info(cluster) | |
| except Exception as e: | |
| logger.error(f"Tool error getting cluster info: {type(e).__name__} - {e}") | |
| # Re-raise to signal failure to the MCP framework/caller | |
| raise | |
| def get_cluster_from_context(ctx: Context) -> Cluster: | |
| cluster = ctx.request_context.lifespan_context.cluster | |
| if not cluster: | |
| raise ValueError("Cluster connection not available in context.") | |
| return cluster | |
| @mcp.tool() | |
| def get_cluster_info(ctx: Context) -> Dict[str, Any]: | |
| """Get diagnostic information about the Couchbase cluster.""" | |
| cluster = get_cluster_from_context(ctx) | |
| try: | |
| return _get_cluster_info(cluster) | |
| except Exception as e: | |
| logger.error(f"Tool error getting cluster info: {type(e).__name__} - {e}") | |
| # Re-raise to signal failure to the MCP framework/caller | |
| raise |
| def get_bucket_info(ctx: Context) -> Dict[str, Any]: | ||
| """Get configuration settings for the current Couchbase bucket.""" | ||
| bucket = ctx.request_context.lifespan_context.bucket | ||
| if not bucket: | ||
| raise ValueError("Bucket connection not available in context.") | ||
| try: | ||
| return _get_bucket_info(bucket) | ||
| except Exception as e: | ||
| logger.error(f"Tool error getting bucket info: {type(e).__name__} - {e}") | ||
| raise | ||
|
|
There was a problem hiding this comment.
Consider extracting the bucket availability check and exception handling into a reusable helper function, as this pattern is repeated in multiple tool functions. This would reduce code duplication and improve maintainability. For example, you could have a function that takes a context and returns the bucket, raising an exception if it's not available, and then use that function in each tool.
| def get_bucket_info(ctx: Context) -> Dict[str, Any]: | |
| """Get configuration settings for the current Couchbase bucket.""" | |
| bucket = ctx.request_context.lifespan_context.bucket | |
| if not bucket: | |
| raise ValueError("Bucket connection not available in context.") | |
| try: | |
| return _get_bucket_info(bucket) | |
| except Exception as e: | |
| logger.error(f"Tool error getting bucket info: {type(e).__name__} - {e}") | |
| raise | |
| def get_bucket_from_context(ctx: Context) -> Any: | |
| bucket = ctx.request_context.lifespan_context.bucket | |
| if not bucket: | |
| raise ValueError("Bucket connection not available in context.") | |
| return bucket | |
| @mcp.tool() | |
| def get_bucket_info(ctx: Context) -> Dict[str, Any]: | |
| """Get configuration settings for the current Couchbase bucket.""" | |
| bucket = get_bucket_from_context(ctx) | |
| try: | |
| return _get_bucket_info(bucket) | |
| except Exception as e: | |
| logger.error(f"Tool error getting bucket info: {type(e).__name__} - {e}") | |
| raise |
| except CouchbaseException as e: | ||
| meta_logger.error(f"Error getting cluster info: {type(e).__name__} - {e}") | ||
| raise # Re-raise for the tool wrapper to handle |
There was a problem hiding this comment.
It's good that you're catching CouchbaseException and logging it. Consider adding more context to the log message, such as the specific operation that failed. Also, ensure that the re-raised exception preserves the original traceback for debugging purposes.
| except CouchbaseException as e: | |
| meta_logger.error(f"Error getting cluster info: {type(e).__name__} - {e}") | |
| raise # Re-raise for the tool wrapper to handle | |
| meta_logger.error(f"Error getting cluster info: {type(e).__name__} - {e}", exc_info=True) | |
| raise |
| except CouchbaseException as e: | ||
| meta_logger.error(f"Error getting bucket info for {bucket.name}: {type(e).__name__} - {e}") | ||
| raise |
There was a problem hiding this comment.
It's good that you're catching CouchbaseException and logging it. Consider adding more context to the log message, such as the specific operation that failed. Also, ensure that the re-raised exception preserves the original traceback for debugging purposes.
| except CouchbaseException as e: | |
| meta_logger.error(f"Error getting bucket info for {bucket.name}: {type(e).__name__} - {e}") | |
| raise | |
| meta_logger.error(f"Error getting bucket info for {bucket.name}: {type(e).__name__} - {e}", exc_info=True) | |
| raise |
| except CouchbaseException as e: | ||
| meta_logger.error(f"Error listing FTS indexes: {type(e).__name__} - {e}") | ||
| raise |
There was a problem hiding this comment.
It's good that you're catching CouchbaseException and logging it. Consider adding more context to the log message, such as the specific operation that failed. Also, ensure that the re-raised exception preserves the original traceback for debugging purposes.
| except CouchbaseException as e: | |
| meta_logger.error(f"Error listing FTS indexes: {type(e).__name__} - {e}") | |
| raise | |
| meta_logger.error(f"Error listing FTS indexes: {type(e).__name__} - {e}", exc_info=True) | |
| raise |
| except CouchbaseException as e: | ||
| meta_logger.error(f"Error listing N1QL indexes for bucket {bucket_name}: {type(e).__name__} - {e}") | ||
| raise |
There was a problem hiding this comment.
It's good that you're catching CouchbaseException and logging it. Consider adding more context to the log message, such as the specific operation that failed. Also, ensure that the re-raised exception preserves the original traceback for debugging purposes.
| except CouchbaseException as e: | |
| meta_logger.error(f"Error listing N1QL indexes for bucket {bucket_name}: {type(e).__name__} - {e}") | |
| raise | |
| meta_logger.error(f"Error listing N1QL indexes for bucket {bucket_name}: {type(e).__name__} - {e}", exc_info=True) | |
| raise |
