Added gRPC gnmi protocol to UTCP #82
Conversation
…tool-calling-protocol/dev Add docs and update http to 1.0.2
…tool-calling-protocol/dev Fix response json parsing when content type is wrong
…om universal-tool-calling-protocol/dev Update CLI
…tool-calling-protocol/dev Update docs
…tool-calling-protocol/dev Plugin updates
…tool-calling-protocol/dev Add WebSocket transport implementation for real-time communication …
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
This PR adds a gNMI (gRPC Network Management Interface) communication protocol plugin to UTCP 1.0, alongside improvements to existing protocol plugins (GraphQL, TCP/UDP sockets) and dependency updates. The gNMI plugin enables network device management via gRPC, implementing discovery, unary calls (capabilities/get/set), streaming subscriptions, and authentication compatible with UTCP's architecture.
Key Changes:
- New gNMI protocol plugin with support for TLS, OAuth2/Basic/API key auth, and streaming subscriptions
- GraphQL plugin refactored to UTCP 1.0 architecture with proper registration and improved header handling
- Socket protocols enhanced with better exception handling, delimiter escape sequence control, and error validation
- Dependency relaxation for MCP plugin's langchain requirement
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 15 comments.
Show a summary per file
| File | Description |
|---|---|
plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py |
Core gNMI protocol implementation with gRPC channel management, metadata/auth handling, and operation routing |
plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_call_template.py |
gNMI call template and serializer definitions |
plugins/communication_protocols/gnmi/src/utcp_gnmi/__init__.py |
Plugin registration entry point |
plugins/communication_protocols/gnmi/tests/test_gnmi_plugin.py |
Tests for manual registration and serializer roundtrip |
plugins/communication_protocols/gnmi/pyproject.toml |
Package configuration with gRPC/protobuf dependencies |
plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py |
Refactored GraphQL protocol to UTCP 1.0 with improved auth and header field mapping |
plugins/communication_protocols/gql/src/utcp_gql/gql_call_template.py |
Added auth serialization/validation to GraphQL provider |
plugins/communication_protocols/gql/src/utcp_gql/__init__.py |
Plugin registration for GraphQL protocol |
plugins/communication_protocols/gql/tests/test_graphql_protocol.py |
New comprehensive tests for GraphQL protocol registration and tool calling |
plugins/communication_protocols/gql/README.md |
Documentation for GraphQL plugin usage |
plugins/communication_protocols/socket/src/utcp_socket/tcp_communication_protocol.py |
Enhanced delimiter handling with escape sequence interpretation flag and unknown framing strategy validation |
plugins/communication_protocols/socket/src/utcp_socket/tcp_call_template.py |
Added interpret_escape_sequences field for delimiter configuration |
plugins/communication_protocols/socket/src/utcp_socket/udp_communication_protocol.py |
Improved exception handling with specific exception types and logging |
plugins/communication_protocols/socket/tests/test_tcp_communication_protocol.py |
Added explanatory comments for exception handling in test server |
plugins/communication_protocols/mcp/pyproject.toml |
Relaxed langchain dependency from exact pin to range constraint |
scripts/socket_sanity.py |
Added explanatory comments for exception handling and updated success message |
Comments suppressed due to low confidence (1)
plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:144
- Variable mode is not used.
mode = tool_args.get("mode", "stream").upper()
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| from utcp.data.auth_implementations.basic_auth import BasicAuth | ||
| from utcp.data.auth_implementations.oauth2_auth import OAuth2Auth | ||
|
|
||
| class GnmiCommunicationProtocol(CommunicationProtocol): |
There was a problem hiding this comment.
The class is missing an __init__ method to initialize the _oauth_tokens dictionary that would be used for OAuth2 token caching (referenced in comment ID 006). Without this initialization, the _handle_oauth2 method will fail with an AttributeError when it tries to access self._oauth_tokens. Add an __init__ method that initializes self._oauth_tokens = {}.
| class GnmiCommunicationProtocol(CommunicationProtocol): | |
| class GnmiCommunicationProtocol(CommunicationProtocol): | |
| def __init__(self): | |
| super().__init__() | |
| self._oauth_tokens = {} |
| { name = "UTCP Contributors" }, | ||
| ] | ||
| description = "UTCP gNMI communication protocol plugin over gRPC" | ||
| readme = "README.md" |
There was a problem hiding this comment.
The pyproject.toml references a README.md file that does not exist in the gnmi plugin directory. This will cause packaging issues. Either create a README.md file with appropriate documentation (similar to the GraphQL plugin's README) or remove the readme field from the project configuration.
| readme = "README.md" |
| async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> Any: | ||
| if not isinstance(tool_call_template, GnmiCallTemplate): | ||
| raise ValueError("GnmiCommunicationProtocol can only be used with GnmiCallTemplate") | ||
|
|
||
| op = tool_call_template.operation | ||
| target = tool_call_template.target | ||
|
|
||
| metadata: List[tuple[str, str]] = [] | ||
| if tool_call_template.metadata: | ||
| metadata.extend([(k, v) for k, v in tool_call_template.metadata.items()]) | ||
| if tool_call_template.metadata_fields: | ||
| for k in tool_call_template.metadata_fields: | ||
| if k in tool_args: | ||
| metadata.append((k, str(tool_args[k]))) | ||
| if tool_call_template.auth: | ||
| if isinstance(tool_call_template.auth, ApiKeyAuth): | ||
| if tool_call_template.auth.api_key: | ||
| metadata.append((tool_call_template.auth.var_name or "authorization", tool_call_template.auth.api_key)) | ||
| elif isinstance(tool_call_template.auth, BasicAuth): | ||
| import base64 | ||
| token = base64.b64encode(f"{tool_call_template.auth.username}:{tool_call_template.auth.password}".encode()).decode() | ||
| metadata.append(("authorization", f"Basic {token}")) | ||
| elif isinstance(tool_call_template.auth, OAuth2Auth): | ||
| token = await self._handle_oauth2(tool_call_template.auth) | ||
| metadata.append(("authorization", f"Bearer {token}")) | ||
|
|
||
| grpc = importlib.import_module("grpc") | ||
| aio = importlib.import_module("grpc.aio") | ||
| json_format = importlib.import_module("google.protobuf.json_format") | ||
| stub_mod = importlib.import_module(tool_call_template.stub_module) | ||
| msg_mod = importlib.import_module(tool_call_template.message_module) | ||
|
|
||
| if tool_call_template.use_tls: | ||
| creds = grpc.ssl_channel_credentials() | ||
| channel = aio.secure_channel(target, creds) | ||
| else: | ||
| channel = aio.insecure_channel(target) | ||
|
|
||
| stub = None | ||
| for attr in dir(stub_mod): | ||
| if attr.endswith("Stub"): | ||
| stub_cls = getattr(stub_mod, attr) | ||
| stub = stub_cls(channel) | ||
| break | ||
| if stub is None: | ||
| raise ValueError("gNMI stub not found in stub_module") | ||
|
|
||
| if op == "capabilities": | ||
| req = getattr(msg_mod, "CapabilityRequest")() | ||
| resp = await stub.Capabilities(req, metadata=metadata) | ||
| elif op == "get": | ||
| req = getattr(msg_mod, "GetRequest")() | ||
| paths = tool_args.get("paths", []) | ||
| for p in paths: | ||
| path_msg = getattr(msg_mod, "Path")() | ||
| for elem in [e for e in p.strip("/").split("/") if e]: | ||
| pe = getattr(msg_mod, "PathElem")(name=elem) | ||
| path_msg.elem.append(pe) | ||
| req.path.append(path_msg) | ||
| resp = await stub.Get(req, metadata=metadata) | ||
| elif op == "set": | ||
| req = getattr(msg_mod, "SetRequest")() | ||
| updates = tool_args.get("updates", []) | ||
| for upd in updates: | ||
| path_msg = getattr(msg_mod, "Path")() | ||
| for elem in [e for e in str(upd.get("path", "")).strip("/").split("/") if e]: | ||
| pe = getattr(msg_mod, "PathElem")(name=elem) | ||
| path_msg.elem.append(pe) | ||
| val = getattr(msg_mod, "TypedValue")(json_val=str(upd.get("value", ""))) | ||
| update_msg = getattr(msg_mod, "Update")(path=path_msg, val=val) | ||
| req.update.append(update_msg) | ||
| resp = await stub.Set(req, metadata=metadata) | ||
| elif op == "subscribe": | ||
| req = getattr(msg_mod, "SubscribeRequest")() | ||
| sub_list = getattr(msg_mod, "SubscriptionList")() | ||
| mode = tool_args.get("mode", "stream").upper() | ||
| sub_list.mode = getattr(msg_mod, "SubscriptionList".upper(), None) or 0 | ||
| paths = tool_args.get("paths", []) | ||
| for p in paths: | ||
| path_msg = getattr(msg_mod, "Path")() | ||
| for elem in [e for e in p.strip("/").split("/") if e]: | ||
| pe = getattr(msg_mod, "PathElem")(name=elem) | ||
| path_msg.elem.append(pe) | ||
| sub = getattr(msg_mod, "Subscription")(path=path_msg) | ||
| sub_list.subscription.append(sub) | ||
| req.subscribe.CopyFrom(sub_list) | ||
| raise ValueError("Subscribe is streaming; use call_tool_streaming") | ||
| else: | ||
| raise ValueError("Unsupported gNMI operation") | ||
|
|
||
| return json_format.MessageToDict(resp) |
There was a problem hiding this comment.
The gRPC channel created on lines 101-105 is never closed. gRPC channels should be properly closed after use to avoid resource leaks. Consider wrapping the channel usage in a try-finally block or using async context manager pattern to ensure the channel is closed.
| async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> AsyncGenerator[Any, None]: | ||
| if not isinstance(tool_call_template, GnmiCallTemplate): | ||
| raise ValueError("GnmiCommunicationProtocol can only be used with GnmiCallTemplate") | ||
| if tool_call_template.operation != "subscribe": | ||
| result = await self.call_tool(caller, tool_name, tool_args, tool_call_template) | ||
| yield result | ||
| return | ||
| grpc = importlib.import_module("grpc") | ||
| aio = importlib.import_module("grpc.aio") | ||
| json_format = importlib.import_module("google.protobuf.json_format") | ||
| stub_mod = importlib.import_module(tool_call_template.stub_module) | ||
| msg_mod = importlib.import_module(tool_call_template.message_module) | ||
| target = tool_call_template.target | ||
| if tool_call_template.use_tls: | ||
| creds = grpc.ssl_channel_credentials() | ||
| channel = aio.secure_channel(target, creds) | ||
| else: | ||
| channel = aio.insecure_channel(target) | ||
| stub = None | ||
| for attr in dir(stub_mod): | ||
| if attr.endswith("Stub"): | ||
| stub_cls = getattr(stub_mod, attr) | ||
| stub = stub_cls(channel) | ||
| break | ||
| if stub is None: | ||
| raise ValueError("gNMI stub not found in stub_module") | ||
| metadata: List[tuple[str, str]] = [] | ||
| if tool_call_template.metadata: | ||
| metadata.extend([(k, v) for k, v in tool_call_template.metadata.items()]) | ||
| if tool_call_template.metadata_fields: | ||
| for k in tool_call_template.metadata_fields: | ||
| if k in tool_args: | ||
| metadata.append((k, str(tool_args[k]))) | ||
| if tool_call_template.auth: | ||
| if isinstance(tool_call_template.auth, ApiKeyAuth): | ||
| if tool_call_template.auth.api_key: | ||
| metadata.append((tool_call_template.auth.var_name or "authorization", tool_call_template.auth.api_key)) | ||
| elif isinstance(tool_call_template.auth, BasicAuth): | ||
| import base64 | ||
| token = base64.b64encode(f"{tool_call_template.auth.username}:{tool_call_template.auth.password}".encode()).decode() | ||
| metadata.append(("authorization", f"Basic {token}")) | ||
| elif isinstance(tool_call_template.auth, OAuth2Auth): | ||
| token = await self._handle_oauth2(tool_call_template.auth) | ||
| metadata.append(("authorization", f"Bearer {token}")) | ||
| req = getattr(msg_mod, "SubscribeRequest")() | ||
| sub_list = getattr(msg_mod, "SubscriptionList")() | ||
| paths = tool_args.get("paths", []) | ||
| for p in paths: | ||
| path_msg = getattr(msg_mod, "Path")() | ||
| for elem in [e for e in p.strip("/").split("/") if e]: | ||
| pe = getattr(msg_mod, "PathElem")(name=elem) | ||
| path_msg.elem.append(pe) | ||
| sub = getattr(msg_mod, "Subscription")(path=path_msg) | ||
| sub_list.subscription.append(sub) | ||
| req.subscribe.CopyFrom(sub_list) | ||
| call = stub.Subscribe(req, metadata=metadata) | ||
| async for resp in call: | ||
| yield json_format.MessageToDict(resp) |
There was a problem hiding this comment.
The gRPC channel created on lines 174-178 is never closed. gRPC channels should be properly closed after use to avoid resource leaks. Consider wrapping the channel usage in a try-finally block or using async context manager pattern to ensure the channel is closed even if the streaming operation is interrupted.
| req = getattr(msg_mod, "SubscribeRequest")() | ||
| sub_list = getattr(msg_mod, "SubscriptionList")() | ||
| mode = tool_args.get("mode", "stream").upper() | ||
| sub_list.mode = getattr(msg_mod, "SubscriptionList".upper(), None) or 0 |
There was a problem hiding this comment.
The getattr(msg_mod, "SubscriptionList".upper(), None) is incorrect and will always evaluate to None. It attempts to get an attribute named "SUBSCRIPTIONLIST" (all uppercase string literal) instead of the subscription mode enum value. This should likely retrieve the subscription mode from the message module based on the mode variable, such as getattr(msg_mod, f"SubscriptionList.Mode.{mode}", 0) or similar, depending on the protobuf structure.
| sub_list.mode = getattr(msg_mod, "SubscriptionList".upper(), None) or 0 | |
| # Set the mode enum value correctly | |
| try: | |
| mode_enum = getattr(getattr(msg_mod, "SubscriptionList"), "Mode") | |
| sub_list.mode = getattr(mode_enum, mode, 0) | |
| except AttributeError: | |
| sub_list.mode = 0 |
| metadata: List[tuple[str, str]] = [] | ||
| if tool_call_template.metadata: | ||
| metadata.extend([(k, v) for k, v in tool_call_template.metadata.items()]) | ||
| if tool_call_template.metadata_fields: | ||
| for k in tool_call_template.metadata_fields: | ||
| if k in tool_args: | ||
| metadata.append((k, str(tool_args[k]))) | ||
| if tool_call_template.auth: | ||
| if isinstance(tool_call_template.auth, ApiKeyAuth): | ||
| if tool_call_template.auth.api_key: | ||
| metadata.append((tool_call_template.auth.var_name or "authorization", tool_call_template.auth.api_key)) | ||
| elif isinstance(tool_call_template.auth, BasicAuth): | ||
| import base64 | ||
| token = base64.b64encode(f"{tool_call_template.auth.username}:{tool_call_template.auth.password}".encode()).decode() | ||
| metadata.append(("authorization", f"Basic {token}")) | ||
| elif isinstance(tool_call_template.auth, OAuth2Auth): | ||
| token = await self._handle_oauth2(tool_call_template.auth) | ||
| metadata.append(("authorization", f"Bearer {token}")) |
There was a problem hiding this comment.
The metadata construction logic (lines 76-93) is duplicated in both call_tool and call_tool_streaming (lines 187-204). This duplication creates a maintenance burden and increases the risk of inconsistencies. Consider extracting this into a private helper method like _build_metadata to eliminate the duplication.
| if manual_call_template.use_tls: | ||
| pass | ||
| else: |
There was a problem hiding this comment.
The empty pass statement in the TLS branch serves no purpose and makes the code harder to read. Consider removing it or adding a comment explaining why TLS is always allowed.
| if manual_call_template.use_tls: | |
| pass | |
| else: | |
| if not manual_call_template.use_tls: |
|
|
||
| async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str: | ||
| import aiohttp | ||
| client_id = auth_details.client_id |
There was a problem hiding this comment.
Variable client_id is not used.
| client_id = auth_details.client_id |
| @@ -0,0 +1,27 @@ | |||
| from typing import Optional, Dict, List, Literal | |||
| from pydantic import Field | |||
There was a problem hiding this comment.
Import of 'Field' is not used.
| from pydantic import Field |
| @@ -0,0 +1,245 @@ | |||
| import importlib | |||
| from typing import Dict, Any, List, Optional, AsyncGenerator | |||
There was a problem hiding this comment.
Import of 'Optional' is not used.
| from typing import Dict, Any, List, Optional, AsyncGenerator | |
| from typing import Dict, Any, List, AsyncGenerator |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
7 issues found across 16 files
Prompt for AI agents (all 7 issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="plugins/communication_protocols/gql/README.md">
<violation number="1" location="plugins/communication_protocols/gql/README.md:12">
P1: Incorrect package name in installation instructions. The package name is `utcp-gql` (as defined in pyproject.toml), not `gql`. The `gql` package is a dependency, not the plugin itself.</violation>
</file>
<file name="plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py">
<violation number="1" location="plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:23">
P1: [gpt-5.2] The localhost check is bypassable via `startswith(...)` (e.g., `localhost.evil.com`). Parse the hostname and compare exact loopback hosts before allowing insecure channels.</violation>
<violation number="2" location="plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:103">
P2: Resource leak: gRPC channel is created but never closed. Consider using `async with` context manager or explicitly calling `await channel.close()` after the RPC call completes to prevent connection leaks.</violation>
<violation number="3" location="plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:137">
P1: [gpt-5.2] TypedValue(json_val=...) is constructed with a Python string, which is not valid for gNMI JSON bytes fields and can break Set RPCs. Serialize values to real JSON bytes (prefer json_ietf_val when available) or fall back to string_val.</violation>
<violation number="4" location="plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:145">
P1: [gpt-5.2] Subscribe `mode` is effectively ignored due to an incorrect `getattr(...)` target; this will likely always set mode to 0. Map the requested mode to the generated enum constant (with a safe default).</violation>
</file>
<file name="plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py">
<violation number="1" location="plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py:193">
P2: [gpt-5.2] Declaring every variable as `String` can break calls for non-String argument types (ID/Int/Boolean/input objects). Infer variable types from the fetched schema (or otherwise avoid hard-coding String).</violation>
<violation number="2" location="plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py:210">
P2: [gpt-5.2] call_tool_streaming currently isn’t truly streaming (it yields once). At minimum, fail fast for subscription operation_type or implement real GraphQL subscription streaming.</violation>
</file>
Reply to cubic to teach it or ask questions. Re-run a review with @cubic-dev-ai review this PR
|
|
||
| ```bash | ||
| pip install gql | ||
| ``` |
There was a problem hiding this comment.
P1: Incorrect package name in installation instructions. The package name is utcp-gql (as defined in pyproject.toml), not gql. The gql package is a dependency, not the plugin itself.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gql/README.md, line 12:
<comment>Incorrect package name in installation instructions. The package name is `utcp-gql` (as defined in pyproject.toml), not `gql`. The `gql` package is a dependency, not the plugin itself.</comment>
<file context>
@@ -1 +1,47 @@
+
+```bash
+pip install gql
+```
+
+### Registration
</file context>
| ``` | |
| pip install utcp-gql |
✅ Addressed in 70ff230
|
|
||
| if tool_call_template.use_tls: | ||
| creds = grpc.ssl_channel_credentials() | ||
| channel = aio.secure_channel(target, creds) |
There was a problem hiding this comment.
P2: Resource leak: gRPC channel is created but never closed. Consider using async with context manager or explicitly calling await channel.close() after the RPC call completes to prevent connection leaks.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py, line 103:
<comment>Resource leak: gRPC channel is created but never closed. Consider using `async with` context manager or explicitly calling `await channel.close()` after the RPC call completes to prevent connection leaks.</comment>
<file context>
@@ -0,0 +1,245 @@
+
+ if tool_call_template.use_tls:
+ creds = grpc.ssl_channel_credentials()
+ channel = aio.secure_channel(target, creds)
+ else:
+ channel = aio.insecure_channel(target)
</file context>
✅ Addressed in 70ff230
| tool_args: Dict[str, Any], | ||
| tool_call_template: CallTemplate, | ||
| ) -> AsyncGenerator[Any, None]: | ||
| # Basic implementation: execute non-streaming and yield once |
There was a problem hiding this comment.
P2: [gpt-5.2] call_tool_streaming currently isn’t truly streaming (it yields once). At minimum, fail fast for subscription operation_type or implement real GraphQL subscription streaming.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py, line 210:
<comment>[gpt-5.2] call_tool_streaming currently isn’t truly streaming (it yields once). At minimum, fail fast for subscription operation_type or implement real GraphQL subscription streaming.</comment>
<file context>
@@ -39,98 +58,158 @@ async def _handle_oauth2(self, auth: OAuth2Auth) -> str:
+ tool_args: Dict[str, Any],
+ tool_call_template: CallTemplate,
+ ) -> AsyncGenerator[Any, None]:
+ # Basic implementation: execute non-streaming and yield once
+ result = await self.call_tool(caller, tool_name, tool_args, tool_call_template)
+ yield result
</file context>
✅ Addressed in 70ff230
| arg_str = ", ".join(f"${k}: String" for k in filtered_args.keys()) | ||
| var_defs = f"({arg_str})" if arg_str else "" | ||
| arg_pass = ', '.join(f"{k}: ${k}" for k in tool_args.keys()) | ||
| arg_pass = ", ".join(f"{k}: ${k}" for k in filtered_args.keys()) | ||
| arg_pass = f"({arg_pass})" if arg_pass else "" |
There was a problem hiding this comment.
P2: [gpt-5.2] Declaring every variable as String can break calls for non-String argument types (ID/Int/Boolean/input objects). Infer variable types from the fetched schema (or otherwise avoid hard-coding String).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py, line 193:
<comment>[gpt-5.2] Declaring every variable as `String` can break calls for non-String argument types (ID/Int/Boolean/input objects). Infer variable types from the fetched schema (or otherwise avoid hard-coding String).</comment>
<file context>
@@ -39,98 +58,158 @@ async def _handle_oauth2(self, auth: OAuth2Auth) -> str:
+ header_fields = tool_call_template.header_fields or []
+ filtered_args = {k: v for k, v in tool_args.items() if k not in header_fields}
+
+ arg_str = ", ".join(f"${k}: String" for k in filtered_args.keys())
var_defs = f"({arg_str})" if arg_str else ""
- arg_pass = ', '.join(f"{k}: ${k}" for k in tool_args.keys())
</file context>
| arg_str = ", ".join(f"${k}: String" for k in filtered_args.keys()) | |
| var_defs = f"({arg_str})" if arg_str else "" | |
| arg_pass = ', '.join(f"{k}: ${k}" for k in tool_args.keys()) | |
| arg_pass = ", ".join(f"{k}: ${k}" for k in filtered_args.keys()) | |
| arg_pass = f"({arg_pass})" if arg_pass else "" | |
| schema = session.client.schema | |
| root_type = { | |
| "query": schema.query_type, | |
| "mutation": schema.mutation_type, | |
| "subscription": schema.subscription_type, | |
| }.get(op_type) | |
| if root_type is None or base_tool_name not in root_type.fields: | |
| raise ValueError(f"Tool '{base_tool_name}' not found on GraphQL {op_type} root type") | |
| field_def = root_type.fields[base_tool_name] | |
| var_def_parts = [] | |
| for k in filtered_args.keys(): | |
| gql_type = str(field_def.args[k].type) if k in field_def.args else "String" | |
| var_def_parts.append(f"${k}: {gql_type}") | |
| var_defs = f"({', '.join(var_def_parts)})" if var_def_parts else "" | |
| arg_pass_parts = [f"{k}: ${k}" for k in filtered_args.keys()] | |
| arg_pass = f"({', '.join(arg_pass_parts)})" if arg_pass_parts else "" |
✅ Addressed in 70ff230
| if manual_call_template.use_tls: | ||
| pass | ||
| else: | ||
| if not (target.startswith("localhost") or target.startswith("127.0.0.1")): |
There was a problem hiding this comment.
P1: [gpt-5.2] The localhost check is bypassable via startswith(...) (e.g., localhost.evil.com). Parse the hostname and compare exact loopback hosts before allowing insecure channels.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py, line 23:
<comment>[gpt-5.2] The localhost check is bypassable via `startswith(...)` (e.g., `localhost.evil.com`). Parse the hostname and compare exact loopback hosts before allowing insecure channels.</comment>
<file context>
@@ -0,0 +1,245 @@
+ if manual_call_template.use_tls:
+ pass
+ else:
+ if not (target.startswith("localhost") or target.startswith("127.0.0.1")):
+ return RegisterManualResult(
+ success=False,
</file context>
✅ Addressed in 70ff230
| req = getattr(msg_mod, "SubscribeRequest")() | ||
| sub_list = getattr(msg_mod, "SubscriptionList")() | ||
| mode = tool_args.get("mode", "stream").upper() | ||
| sub_list.mode = getattr(msg_mod, "SubscriptionList".upper(), None) or 0 |
There was a problem hiding this comment.
P1: [gpt-5.2] Subscribe mode is effectively ignored due to an incorrect getattr(...) target; this will likely always set mode to 0. Map the requested mode to the generated enum constant (with a safe default).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py, line 145:
<comment>[gpt-5.2] Subscribe `mode` is effectively ignored due to an incorrect `getattr(...)` target; this will likely always set mode to 0. Map the requested mode to the generated enum constant (with a safe default).</comment>
<file context>
@@ -0,0 +1,245 @@
+ req = getattr(msg_mod, "SubscribeRequest")()
+ sub_list = getattr(msg_mod, "SubscriptionList")()
+ mode = tool_args.get("mode", "stream").upper()
+ sub_list.mode = getattr(msg_mod, "SubscriptionList".upper(), None) or 0
+ paths = tool_args.get("paths", [])
+ for p in paths:
</file context>
✅ Addressed in 70ff230
| for elem in [e for e in str(upd.get("path", "")).strip("/").split("/") if e]: | ||
| pe = getattr(msg_mod, "PathElem")(name=elem) | ||
| path_msg.elem.append(pe) | ||
| val = getattr(msg_mod, "TypedValue")(json_val=str(upd.get("value", ""))) |
There was a problem hiding this comment.
P1: [gpt-5.2] TypedValue(json_val=...) is constructed with a Python string, which is not valid for gNMI JSON bytes fields and can break Set RPCs. Serialize values to real JSON bytes (prefer json_ietf_val when available) or fall back to string_val.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py, line 137:
<comment>[gpt-5.2] TypedValue(json_val=...) is constructed with a Python string, which is not valid for gNMI JSON bytes fields and can break Set RPCs. Serialize values to real JSON bytes (prefer json_ietf_val when available) or fall back to string_val.</comment>
<file context>
@@ -0,0 +1,245 @@
+ for elem in [e for e in str(upd.get("path", "")).strip("/").split("/") if e]:
+ pe = getattr(msg_mod, "PathElem")(name=elem)
+ path_msg.elem.append(pe)
+ val = getattr(msg_mod, "TypedValue")(json_val=str(upd.get("value", "")))
+ update_msg = getattr(msg_mod, "Update")(path=path_msg, val=val)
+ req.update.append(update_msg)
</file context>
✅ Addressed in 70ff230
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 17 out of 17 changed files in this pull request and generated 10 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| class GnmiCommunicationProtocol(CommunicationProtocol): | ||
| def __init__(self): | ||
| self._oauth_tokens: Dict[str, Dict[str, Any]] = {} |
There was a problem hiding this comment.
The GnmiCommunicationProtocol class is missing a close() method. The GraphQL protocol implements this method (line 228-229 in gql_communication_protocol.py) to clean up OAuth tokens. For consistency and proper resource cleanup, the gNMI protocol should also implement this method.
| except Exception: | ||
| from aiohttp import BasicAuth as AiohttpBasicAuth | ||
| header_auth = AiohttpBasicAuth(auth_details.client_id, auth_details.client_secret) |
There was a problem hiding this comment.
The bare except Exception: clause silently catches all exceptions from the first OAuth2 attempt and tries an alternative authentication method. This could mask programming errors or configuration issues. Consider catching specific exceptions (e.g., aiohttp.ClientResponseError) and logging the failure before attempting the fallback method. This will help with debugging when the primary authentication method fails.
| def _load_gnmi_modules(self, tool_call_template: GnmiCallTemplate): | ||
| grpc = importlib.import_module("grpc") | ||
| aio = importlib.import_module("grpc.aio") | ||
| json_format = importlib.import_module("google.protobuf.json_format") | ||
| stub_mod = importlib.import_module(tool_call_template.stub_module) | ||
| msg_mod = importlib.import_module(tool_call_template.message_module) | ||
| return grpc, aio, json_format, stub_mod, msg_mod |
There was a problem hiding this comment.
The _load_gnmi_modules method lacks return type annotations. Consider adding the return type annotation for better type safety and IDE support. For example: -> tuple[Any, Any, Any, Any, Any] or more specifically defining the types of grpc, aio, json_format, stub_mod, and msg_mod.
| def _create_grpc_channel(self, grpc, aio, target: str, use_tls: bool): | ||
| if use_tls: | ||
| creds = grpc.ssl_channel_credentials() | ||
| return aio.secure_channel(target, creds) | ||
| return aio.insecure_channel(target) | ||
| def _create_grpc_stub(self, stub_mod, channel): | ||
| stub = None | ||
| for attr in dir(stub_mod): | ||
| if attr.endswith("Stub"): | ||
| stub_cls = getattr(stub_mod, attr) | ||
| stub = stub_cls(channel) | ||
| break | ||
| if stub is None: | ||
| raise ValueError("gNMI stub not found in stub_module") | ||
| return stub |
There was a problem hiding this comment.
The _create_grpc_channel and _create_grpc_stub methods lack return type annotations. Adding type hints would improve code clarity and enable better static analysis.
| async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> Any: | ||
| if not isinstance(tool_call_template, GnmiCallTemplate): | ||
| raise ValueError("GnmiCommunicationProtocol can only be used with GnmiCallTemplate") | ||
|
|
||
| op = tool_call_template.operation | ||
| target = tool_call_template.target | ||
|
|
||
| metadata: List[tuple[str, str]] = [] | ||
| if tool_call_template.metadata: | ||
| metadata.extend([(k, v) for k, v in tool_call_template.metadata.items()]) | ||
| if tool_call_template.metadata_fields: | ||
| for k in tool_call_template.metadata_fields: | ||
| if k in tool_args: | ||
| metadata.append((k, str(tool_args[k]))) | ||
| if tool_call_template.auth: | ||
| if isinstance(tool_call_template.auth, ApiKeyAuth): | ||
| if tool_call_template.auth.api_key: | ||
| metadata.append((tool_call_template.auth.var_name or "authorization", tool_call_template.auth.api_key)) | ||
| elif isinstance(tool_call_template.auth, BasicAuth): | ||
| import base64 | ||
| token = base64.b64encode(f"{tool_call_template.auth.username}:{tool_call_template.auth.password}".encode()).decode() | ||
| metadata.append(("authorization", f"Basic {token}")) | ||
| elif isinstance(tool_call_template.auth, OAuth2Auth): | ||
| token = await self._handle_oauth2(tool_call_template.auth) | ||
| metadata.append(("authorization", f"Bearer {token}")) |
There was a problem hiding this comment.
The authentication metadata building logic is duplicated between call_tool (lines 109-126) and call_tool_streaming (lines 197-214). Consider extracting this into a shared helper method like _build_metadata(tool_call_template, tool_args) to reduce code duplication and improve maintainability. This pattern is already used in the GraphQL protocol with _prepare_headers().
| raise ValueError("Unsupported gNMI operation") | ||
| else: | ||
| raise ValueError("Unsupported gNMI operation") |
There was a problem hiding this comment.
The error message "Unsupported gNMI operation" is generic and doesn't indicate what operation was provided. Consider including the actual operation value in the error message for easier debugging, e.g., f"Unsupported gNMI operation: {op}".
| raise ValueError("Unsupported gNMI operation") | |
| else: | |
| raise ValueError("Unsupported gNMI operation") | |
| raise ValueError(f"Unsupported gNMI operation: {op}") | |
| else: | |
| raise ValueError(f"Unsupported gNMI operation: {op}") |
plugins/communication_protocols/socket/src/utcp_socket/tcp_communication_protocol.py
Show resolved
Hide resolved
| except (UtcpSerializerValidationError, ValueError) as e: | ||
| # Fallback to manual template if validation fails, but log details | ||
| logger.exception("Failed to validate existing tool_call_template; falling back to manual template") | ||
| normalized["tool_call_template"] = manual_call_template |
There was a problem hiding this comment.
While catching specific exception types (UtcpSerializerValidationError, ValueError) is better than a bare except Exception, you should not suppress the variable e in the except clause. The caught exception is logged via logger.exception() which will include the traceback, but explicitly using the exception variable would be more clear. Also, the except clause on line 112 is inconsistent - it only catches UtcpSerializerValidationError, not ValueError. Consider standardizing the exception handling across both branches.
| class GnmiCallTemplate(CallTemplate): | ||
| call_template_type: Literal["gnmi"] = "gnmi" | ||
| target: str | ||
| use_tls: bool = True | ||
| metadata: Optional[Dict[str, str]] = None | ||
| metadata_fields: Optional[List[str]] = None | ||
| operation: Literal["capabilities", "get", "set", "subscribe"] = "get" | ||
| stub_module: str = "gnmi_pb2_grpc" | ||
| message_module: str = "gnmi_pb2" |
There was a problem hiding this comment.
The GnmiCallTemplate class is missing auth serialization/validation similar to what GraphQLProvider has. GraphQLProvider includes @field_serializer("auth") and @field_validator("auth", mode="before") decorators (lines 34-47 in gql_call_template.py) to properly handle Auth object serialization and deserialization. Without these, the auth field may not serialize/deserialize correctly when going through the CallTemplate lifecycle.
There was a problem hiding this comment.
@copilot apply changes based on this feedback to the current PR
| @pytest.mark.asyncio | ||
| async def test_register_manual_and_tools(): | ||
| register() | ||
| client = await UtcpClient.create(config={ | ||
| "manual_call_templates": [ | ||
| { | ||
| "name": "routerA", | ||
| "call_template_type": "gnmi", | ||
| "target": "localhost:50051", | ||
| "use_tls": False, | ||
| "operation": "get" | ||
| } | ||
| ] | ||
| }) | ||
| tools = await client.config.tool_repository.get_tools() | ||
| names = [t.name for t in tools] | ||
| assert any(n.startswith("routerA.") for n in names) | ||
| assert any(n.endswith("subscribe") for n in names) |
There was a problem hiding this comment.
The test suite lacks coverage for the core call_tool and call_tool_streaming methods in GnmiCommunicationProtocol. While test_register_manual_and_tools validates tool registration, there are no tests for actual RPC calls (capabilities, get, set) or streaming subscribe. The GraphQL plugin includes comprehensive tests that call tools (lines 103-110 in test_graphql_protocol.py). Consider adding similar tests with mocked gRPC stubs to verify the request building and response handling logic.
There was a problem hiding this comment.
@copilot l request to apply changes based on this feedback to the current PR
|
@h3xxit review these changes and let me know if there is something to update. |
2 participants
This PR adds a gNMI (gRPC) protocol plugin fully compatible with UTCP 1.0 and its protocol/call-template architecture. It implements discovery, calling (unary), streaming subscribe, authentication, registration, and security alignment consistent with other UTCP plugins.
Protocol Implementation
Added GnmiCommunicationProtocol conforming to UTCP’s CommunicationProtocol interface
Auth injection into gRPC metadata
Protobuf request binding
Call Template & Serializer
Introduced GnmiCallTemplate with:
Added GnmiCallTemplateSerializer that validates and serializes the template at plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_call_template.py:19–27
Registration
Registers both protocol and call template via UTCP plugin system:
Packaging entry point configured:
Security & Validation
TLS enforcement unless target is localhost / 127.0.0.1 at plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:16–26
Merges static metadata and whitelisted dynamic metadata_fields into gRPC call metadata at plugins/communication_protocols/gnmi/src/utcp_gnmi/gnmi_communication_protocol.py:73–80
Proper error propagation via UTCP client routing in core/src/utcp/implementations/utcp_client_implementation.py:176–197
Testing
Summary by cubic
Adds a gNMI (gRPC) communication protocol plugin for UTCP 1.0 with secure discovery, unary calls (Capabilities/Get/Set), and streaming Subscribe. Also modernizes GraphQL and socket protocols for UTCP 1.0 compatibility and better reliability.
New Features
Refactors
Written for commit 9bbb819. Summary will update automatically on new commits.