Add network metadata to MCP tool _meta field#769
Add network metadata to MCP tool _meta field#769andreidenissov-cog wants to merge 8 commits intomainfrom
Conversation
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
neuro_san/service/mcp/processors/mcp_tools_processor.py:227
- The type signature says this returns
Dict[str, Any], but the function returnsNonewhen the agent is unauthorized or missing a service provider. Update the return type toOptional[Dict[str, Any]](and propagate that to callers) to avoid misleading type hints and downstreamNonehandling bugs.
async def _get_tool_description(
self,
agent_name: str,
agent_network: AgentNetwork,
metadata: Dict[str, Any]) -> Dict[str, Any]:
is_authorized: bool = False
service_provider: AsyncAgentServiceProvider = None
is_authorized, service_provider = await self.agent_policy.allow_agent(agent_name, metadata)
if service_provider is None or not is_authorized:
return None
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
For MCP list tools response, include the agent network definition metadata in the _meta part of each tool description so clients can access it. Add get_metadata() to AgentNetwork for clean access to the metadata section of the network config. Co-Authored-By: andrei.denissov <andrei.denissov@cognizant.com>
Make sure agent network metadata is a dictionary instance. Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
634f86d to
f997d03
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # List of keys we are interested in for MCP tool metadata - we can add more if necessary: | ||
| metadata_keys: List[str] = ["description", "sample_queries", "tags"] | ||
|
|
||
| network_metadata: Any = agent_network.get_metadata() | ||
| result_dict: Dict[str, Any] = {} | ||
| if isinstance(network_metadata, dict): | ||
| for key in metadata_keys: | ||
| if key in network_metadata: | ||
| result_dict[key] = network_metadata[key] | ||
| if len(result_dict) > 0: |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
There was a problem hiding this comment.
PR description updated.
| async def get_agent_network_metadata(self, agent_network: AgentNetwork) -> Dict[str, Any]: | ||
| """ | ||
| Get agent network metadata dictionary for MCP tool description. | ||
| :param agent_network: agent network object; | ||
| :return: metadata dictionary or None | ||
| """ | ||
| # List of keys we are interested in for MCP tool metadata - we can add more if necessary: | ||
| metadata_keys: List[str] = ["description", "sample_queries", "tags"] | ||
|
|
||
| network_metadata: Any = agent_network.get_metadata() | ||
| result_dict: Dict[str, Any] = {} | ||
| if isinstance(network_metadata, dict): | ||
| for key in metadata_keys: | ||
| if key in network_metadata: | ||
| result_dict[key] = network_metadata[key] | ||
| if len(result_dict) > 0: | ||
| return result_dict | ||
| return None |
There was a problem hiding this comment.
Type annotations don’t match behavior: get_agent_network_metadata() is annotated to return Dict[str, Any] but returns None when there’s no metadata, and callers treat it as optional. Update the return type to Optional[Dict[str, Any]] (and consider making the method non-async since it doesn’t await anything).
| mcp_session: McpServiceAgentSession = self.session | ||
| response_dict: Dict[str, Any] = mcp_session.list({}) | ||
| tools_list: List[Dict[str, Any]] = response_dict.get("result", {}).get("tools", []) | ||
| tools_description: List[Dict[str, Any]] = [] | ||
| for tool in tools_list: | ||
| metadata: Dict[str, str] = tool.get("_meta", {}) | ||
| tool_description: Dict[str, Any] = { | ||
| "agent_name": tool.get("name", "<Unnamed Tool>"), | ||
| "description": metadata.get("description", ""), | ||
| "tags": metadata.get("tags", []) | ||
| } | ||
| tools_description.append(tool_description) | ||
|
|
||
| empty_list: List[Dict[str, Any]] = [] | ||
| if self.args.tags: | ||
| tags = set() | ||
| for tool_info in tools_description: | ||
| tool_tags: List[str] = tool_info.get("tags", empty_list) | ||
| tags.update(tool_tags) |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
| path: str = self.get_request_path("list") | ||
| try: | ||
| with requests.post(path, json=mcp_payload, headers=headers, | ||
| timeout=self.streaming_timeout_in_seconds) as response: |
There was a problem hiding this comment.
list() uses timeout=self.streaming_timeout_in_seconds, which is typically None for MCP sessions created by the CLI/factory, resulting in an unbounded HTTP request for a non-streaming call. Use the regular timeout_in_seconds (or a dedicated short timeout) for tools/list so the CLI doesn’t hang indefinitely when the MCP server is unreachable.
| timeout=self.streaming_timeout_in_seconds) as response: | |
| timeout=self.timeout_in_seconds) as response: |
There was a problem hiding this comment.
Good catch. Fixed.
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (1)
neuro_san/service/mcp/processors/mcp_tools_processor.py:228
_get_tool_description()is declared to returnDict[str, Any], but it returnsNonewhen the agent is not authorized / no provider is found. Please update the return type toOptional[Dict[str, Any]](or refactor to always return a dict/error) so callers and type checkers can handle this correctly.
async def _get_tool_description(
self,
agent_name: str,
agent_network: AgentNetwork,
metadata: Dict[str, Any]) -> Dict[str, Any]:
is_authorized: bool = False
service_provider: AsyncAgentServiceProvider = None
is_authorized, service_provider = await self.agent_policy.allow_agent(agent_name, metadata)
if service_provider is None or not is_authorized:
return None
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| tool_dict: Dict[str, Any] = \ | ||
| await self._get_tool_description(agent_name, agent_network, metadata) | ||
| if tool_dict is not None: | ||
| tools_description.append(tool_dict) |
There was a problem hiding this comment.
tool_dict is annotated as Dict[str, Any] but _get_tool_description() can return None (see its early return). This makes the type hint incorrect and can mask real issues in static analysis; consider annotating tool_dict as Optional[Dict[str, Any]] (and/or make _get_tool_description always return a dict).
| async def get_agent_network_metadata(self, agent_network: AgentNetwork) -> Dict[str, Any]: | ||
| """ | ||
| Get agent network metadata dictionary for MCP tool description. | ||
| :param agent_network: agent network object; | ||
| :return: metadata dictionary or None | ||
| """ | ||
| # List of keys we are interested in for MCP tool metadata - we can add more if necessary: | ||
| metadata_keys: List[str] = ["description", "sample_queries", "tags"] | ||
|
|
||
| network_metadata: Any = agent_network.get_metadata() | ||
| result_dict: Dict[str, Any] = {} | ||
| if isinstance(network_metadata, dict): | ||
| for key in metadata_keys: | ||
| if key in network_metadata: | ||
| result_dict[key] = network_metadata[key] | ||
| if len(result_dict) > 0: | ||
| return result_dict | ||
| return None | ||
|
|
There was a problem hiding this comment.
get_agent_network_metadata() is marked async and annotated to return Dict[str, Any], but it contains no awaits and can return None. Consider making it a regular def and changing the return type to Optional[Dict[str, Any]] (and updating the call site) to better reflect its behavior.
| tool_description: Dict[str, Any] = { | ||
| "agent_name": tool.get("name", "<Unnamed Tool>"), | ||
| "description": metadata.get("description", ""), | ||
| "tags": metadata.get("tags", []) |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
| return | ||
|
|
||
| self.open_session() | ||
|
|
There was a problem hiding this comment.
For MCP mode, we need session to be opened - to execute initial handshake.
There was a problem hiding this comment.
Makes sense — the MCP tools/list call requires the session to be established first (via the initialize / notifications/initialized handshake), so open_session() must run before mcp_list() can send the request.
| tool_description: Dict[str, Any] = { | ||
| "agent_name": tool.get("name", "<Unnamed Tool>"), | ||
| "description": metadata.get("description", ""), | ||
| "tags": metadata.get("tags", []) |
There was a problem hiding this comment.
Extract description and tags from MCP tool metadata, so displayed data will be consistent with what we publish for agent networks.
There was a problem hiding this comment.
Good approach — pulling description and tags from _meta keeps the CLI output aligned with the server-side metadata published in the tools/list response.
| or None if no metadata is defined. | ||
| """ | ||
| return self.config.get("metadata", None) | ||
|
|
There was a problem hiding this comment.
Helper method for getting agent network metadata.
| raise ValueError(self.help_message(path)) from exc | ||
|
|
||
| def list(self, request_dict: Dict[str, Any]) -> Dict[str, Any]: | ||
| """ |
There was a problem hiding this comment.
We extend McpServiceAgentSession with list() API call from ConciergeSession to have all MCP-related functionality provided by a single session, initial handshake and all.
There was a problem hiding this comment.
🔴 "mcp" missing from --connection argparse choices, making --connection mcp fail
The choices parameter for --connection is ["direct", "http", "https"] but does not include "mcp". This means --connection mcp will be rejected by argparse with an error, even though the help text documents it as a valid choice and the rest of the code (main() at line 69) checks for self.args.connection == "mcp". While --mcp (the flag variant) works because store_const bypasses choices validation, the --connection mcp form that is documented in the help text is broken.
(Refers to line 281)
Was this helpful? React with 👍 or 👎 to provide feedback.
| metadata: Dict[str, Any] = tool.get("_meta", {}) | ||
| tool_description: Dict[str, Any] = { | ||
| "agent_name": tool.get("name", "<Unnamed Tool>"), | ||
| "description": metadata.get("description", ""), | ||
| "tags": metadata.get("tags", []) | ||
| } |
There was a problem hiding this comment.
🔴 mcp_list reads description/tags from _meta instead of tool-level fields, showing empty values for tools without HOCON metadata
In mcp_list(), the tool description and tags are read exclusively from the _meta sub-dictionary (tool.get("_meta", {})). However, on the server side (mcp_tools_processor.py:239-241), _meta is only populated when the agent network has a metadata section in its HOCON config. For tools without HOCON metadata, _meta won't be present, so tool.get("_meta", {}) returns {}, and the output will show an empty description and empty tags — even though the tool's top-level description field (populated from the agent's function description at mcp_tools_processor.py:232-235) contains a valid value. The description should fall back to tool.get("description", "") when _meta doesn't contain one.
| metadata: Dict[str, Any] = tool.get("_meta", {}) | |
| tool_description: Dict[str, Any] = { | |
| "agent_name": tool.get("name", "<Unnamed Tool>"), | |
| "description": metadata.get("description", ""), | |
| "tags": metadata.get("tags", []) | |
| } | |
| metadata: Dict[str, Any] = tool.get("_meta", {}) | |
| tool_description: Dict[str, Any] = { | |
| "agent_name": tool.get("name", "<Unnamed Tool>"), | |
| "description": metadata.get("description", tool.get("description", "")), | |
| "tags": metadata.get("tags", []) | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
NS-736 Allow --list option in agent_cli client for MCP connection.
Summary
For MCP
tools/listresponses, the agent network'smetadatasection (from the HOCON definition) is now included in the_metafield of each tool description. Note that this metadata is currently filtered to only include keysdescription,tags, andsample_queries.Changes:
get_metadata()accessor toAgentNetworkto retrieve themetadatadict from the network config_get_tool_descriptioninMcpToolsProcessorto accept theAgentNetworkand populate_metawith the network metadata when presentReview & Testing Checklist for Human
_metaat the tool object level (alongsidename,description,inputSchema) is correct per MCP 2025-06-18 spec for tool annotations_metais the desired behavior — no fields in the HOCONmetadatasection should be filtered or excluded from client visibilityget_metadata()is annotated asDict[str, Any]but can returnNone— consider whether it should beOptional[Dict[str, Any]]tools/listrequest against an agent network that has ametadatasection and verify_metaappears correctly in the response; also test against one withoutmetadatato confirm_metais omittedNotes