Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "utcp"
version = "1.1.0"
version = "1.1.1"
authors = [
{ name = "UTCP Contributors" },
]
Expand Down
35 changes: 25 additions & 10 deletions core/src/utcp/implementations/default_variable_substitutor.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,10 @@ def substitute(self, obj: dict | list | str, config: UtcpClientConfig, variable_
Non-string types are returned unchanged. String values are scanned
for variable references using ${VAR} and $VAR syntax.

Note:
Strings containing '$ref' are skipped to support OpenAPI specs
stored as string content, where $ref is a JSON reference keyword.

Args:
obj: Object to perform substitution on. Can be any type.
config: UTCP client configuration containing variable sources.
Expand Down Expand Up @@ -95,18 +99,22 @@ def substitute(self, obj: dict | list | str, config: UtcpClientConfig, variable_
if variable_namespace and not all(c.isalnum() or c == '_' for c in variable_namespace):
raise ValueError(f"Variable namespace '{variable_namespace}' contains invalid characters. Only alphanumeric characters and underscores are allowed.")

if isinstance(obj, dict):
return {k: self.substitute(v, config, variable_namespace) for k, v in obj.items()}
elif isinstance(obj, list):
return [self.substitute(elem, config, variable_namespace) for elem in obj]
elif isinstance(obj, str):
if isinstance(obj, str):
# Skip substitution for JSON $ref strings
if '$ref' in obj:
Copy link
Contributor

@cubic-dev-ai cubic-dev-ai bot Nov 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Strings such as $refresh_token now bypass substitution entirely because they contain the substring $ref, so legitimate variables that merely start with “ref” are no longer resolved.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At core/src/utcp/implementations/default_variable_substitutor.py, line 104:

<comment>Strings such as `$refresh_token` now bypass substitution entirely because they contain the substring `$ref`, so legitimate variables that merely start with “ref” are no longer resolved.</comment>

<file context>
@@ -95,18 +99,22 @@ def substitute(self, obj: dict | list | str, config: UtcpClientConfig, variable_
-        elif isinstance(obj, str):
+        if isinstance(obj, str):
+            # Skip substitution for JSON $ref strings
+            if &#39;$ref&#39; in obj:
+                return obj
+
</file context>
Suggested change
if '$ref' in obj:
if '"$ref"' in obj:
Fix with Cubic

return obj

# Use a regular expression to find all variables in the string, supporting ${VAR} and $VAR formats
def replacer(match):
# The first group that is not None is the one that matched
var_name = next((g for g in match.groups() if g is not None), "")
return self._get_variable(var_name, config, variable_namespace)

return re.sub(r'\${(\w+)}|\$(\w+)', replacer, obj)
return re.sub(r'\${([a-zA-Z0-9_]+)}|\$([a-zA-Z0-9_]+)', replacer, obj)
elif isinstance(obj, dict):
return {k: self.substitute(v, config, variable_namespace) for k, v in obj.items()}
elif isinstance(obj, list):
return [self.substitute(elem, config, variable_namespace) for elem in obj]
else:
return obj

Expand All @@ -118,6 +126,10 @@ def find_required_variables(self, obj: dict | list | str, variable_namespace: Op
returning fully-qualified variable names with variable namespacing.
Useful for validation and dependency analysis.

Note:
Strings containing '$ref' are skipped to support OpenAPI specs
stored as string content, where $ref is a JSON reference keyword.

Args:
obj: Object to scan for variable references.
variable_namespace: Variable namespace used for variable namespacing.
Expand All @@ -127,7 +139,7 @@ def find_required_variables(self, obj: dict | list | str, variable_namespace: Op
ValueError: If variable_namespace contains invalid characters.

Returns:
List of fully-qualified variable names found in the object.
List of unique fully-qualified variable names found in the object.

Example:
```python
Expand Down Expand Up @@ -156,19 +168,22 @@ def find_required_variables(self, obj: dict | list | str, variable_namespace: Op
result.extend(vars)
return result
elif isinstance(obj, str):
# Skip substitution for JSON $ref strings
if '$ref' in obj:
return []
# Find all variables in the string, supporting ${VAR} and $VAR formats
variables = []
pattern = r'\${(\w+)}|\$(\w+)'
pattern = r'\${([a-zA-Z0-9_]+)}|\$([a-zA-Z0-9_]+)'

for match in re.finditer(pattern, obj):
# The first group that is not None is the one that matched
var_name = next(g for g in match.groups() if g is not None)
if variable_namespace:
full_var_name = variable_namespace.replace("_", "!").replace("!", "__") + "_" + var_name
full_var_name = variable_namespace.replace("_", "__") + "_" + var_name
else:
full_var_name = var_name
variables.append(full_var_name)

return variables
return list(set(variables))
else:
return []
4 changes: 2 additions & 2 deletions plugins/communication_protocols/cli/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "utcp-cli"
version = "1.1.0"
version = "1.1.1"
authors = [
{ name = "UTCP Contributors" },
]
Expand All @@ -14,7 +14,7 @@ requires-python = ">=3.10"
dependencies = [
"pydantic>=2.0",
"pyyaml>=6.0",
"utcp>=1.0"
"utcp>=1.1"
]
classifiers = [
"Development Status :: 4 - Beta",
Expand Down
129 changes: 129 additions & 0 deletions plugins/communication_protocols/file/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
# UTCP File Plugin

[![PyPI Downloads](https://static.pepy.tech/badge/utcp-file)](https://pepy.tech/projects/utcp-file)

A file-based resource plugin for UTCP. This plugin allows you to define tools that return the content of a specified local file.

## Features

- **Local File Content**: Define tools that read and return the content of local files.
- **UTCP Manual Discovery**: Load tool definitions from local UTCP manual files in JSON or YAML format.
- **OpenAPI Support**: Automatically converts local OpenAPI specs to UTCP tools with optional authentication.
- **Static & Simple**: Ideal for returning mock data, configuration, or any static text content from a file.
- **Version Control**: Tool definitions and their corresponding content files can be versioned with your code.
- **No File Authentication**: Designed for simple, local file access without authentication for file reading.
- **Tool Authentication**: Supports authentication for generated tools from OpenAPI specs via `auth_tools`.

## Installation

```bash
pip install utcp-file
```

## How It Works

The File plugin operates in two main ways:

1. **Tool Discovery (`register_manual`)**: It can read a standard UTCP manual file (e.g., `my-tools.json`) to learn about available tools. This is how the `UtcpClient` discovers what tools can be called.
2. **Tool Execution (`call_tool`)**: When you call a tool, the plugin looks at the `tool_call_template` associated with that tool. It expects a `file` template, and it will read and return the entire content of the `file_path` specified in that template.

**Important**: The `call_tool` function **does not** use the arguments you pass to it. It simply returns the full content of the file defined in the tool's template.

## Quick Start

Here is a complete example demonstrating how to define and use a tool that returns the content of a file.

### 1. Create a Content File

First, create a file with some content that you want your tool to return.

`./mock_data/user.json`:
```json
{
"id": 123,
"name": "John Doe",
"email": "john.doe@example.com"
}
```

### 2. Create a UTCP Manual

Next, define a UTCP manual that describes your tool. The `tool_call_template` must be of type `file` and point to the content file you just created.

`./manuals/local_tools.json`:
```json
{
"manual_version": "1.0.0",
"utcp_version": "1.0.2",
"tools": [
{
"name": "get_mock_user",
"description": "Returns a mock user profile from a local file.",
"tool_call_template": {
"call_template_type": "file",
"file_path": "./mock_data/user.json"
}
}
]
}
```

### 3. Use the Tool in Python

Finally, use the `UtcpClient` to load the manual and call the tool.

```python
import asyncio
from utcp.utcp_client import UtcpClient

async def main():
# Create a client, providing the path to the manual.
# The file plugin is used automatically for the "file" call_template_type.
client = await UtcpClient.create(config={
"manual_call_templates": [{
"name": "local_file_tools",
"call_template_type": "file",
"file_path": "./manuals/local_tools.json"
}]
})

# List the tools to confirm it was loaded
tools = await client.list_tools()
print("Available tools:", [tool.name for tool in tools])

# Call the tool. The result will be the content of './mock_data/user.json'
result = await client.call_tool("local_file_tools.get_mock_user", {})

print("\nTool Result:")
print(result)

if __name__ == "__main__":
asyncio.run(main())
```

### Expected Output:

```
Available tools: ['local_file_tools.get_mock_user']

Tool Result:
{
"id": 123,
"name": "John Doe",
"email": "john.doe@example.com"
}
```

## Use Cases

- **Mocking**: Return mock data for tests or local development without needing a live server.
- **Configuration**: Load static configuration files as tool outputs.
- **Templates**: Retrieve text templates (e.g., for emails or reports).

## Related Documentation

- [Main UTCP Documentation](../../../README.md)
- [Core Package Documentation](../../../core/README.md)
- [HTTP Plugin](../http/README.md) - For calling real web APIs.
- [Text Plugin](../text/README.md) - For direct text content (browser-compatible).
- [CLI Plugin](../cli/README.md) - For executing command-line tools.
45 changes: 45 additions & 0 deletions plugins/communication_protocols/file/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "utcp-file"
version = "1.1.0"
authors = [
{ name = "UTCP Contributors" },
]
description = "UTCP communication protocol plugin for reading local files."
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"pydantic>=2.0",
"pyyaml>=6.0",
"utcp>=1.1",
"utcp-http>=1.1",
"aiofiles>=23.2.1"
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Operating System :: OS Independent",
]
license = "MPL-2.0"

[project.optional-dependencies]
dev = [
"build",
"pytest",
"pytest-asyncio",
"pytest-cov",
"coverage",
"twine",
]

[project.urls]
Homepage = "https://utcp.io"
Source = "https://github.com/universal-tool-calling-protocol/python-utcp"
Issues = "https://github.com/universal-tool-calling-protocol/python-utcp/issues"

[project.entry-points."utcp.plugins"]
file = "utcp_file:register"
17 changes: 17 additions & 0 deletions plugins/communication_protocols/file/src/utcp_file/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
"""File Communication Protocol plugin for UTCP."""

from utcp.plugins.discovery import register_communication_protocol, register_call_template
from utcp_file.file_communication_protocol import FileCommunicationProtocol
from utcp_file.file_call_template import FileCallTemplate, FileCallTemplateSerializer


def register():
register_communication_protocol("file", FileCommunicationProtocol())
register_call_template("file", FileCallTemplateSerializer())


__all__ = [
"FileCommunicationProtocol",
"FileCallTemplate",
"FileCallTemplateSerializer",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
from typing import Literal, Optional, Any
from pydantic import Field, field_serializer, field_validator

from utcp.data.call_template import CallTemplate
from utcp.data.auth import Auth, AuthSerializer
from utcp.interfaces.serializer import Serializer
from utcp.exceptions import UtcpSerializerValidationError
import traceback


class FileCallTemplate(CallTemplate):
"""REQUIRED
Call template for file-based manuals and tools.

Reads UTCP manuals or tool definitions from local JSON/YAML files. Useful for
static tool configurations or environments where manuals are distributed as files.
For direct text content, use the text protocol instead.

Attributes:
call_template_type: Always "file" for file call templates.
file_path: Path to the file containing the UTCP manual or tool definitions.
auth: Always None - file call templates don't support authentication for file access.
auth_tools: Optional authentication to apply to generated tools from OpenAPI specs.
"""

call_template_type: Literal["file"] = "file"
file_path: str = Field(..., description="The path to the file containing the UTCP manual or tool definitions.")
auth: None = None
auth_tools: Optional[Auth] = Field(None, description="Authentication to apply to generated tools from OpenAPI specs.")

@field_serializer('auth_tools')
def serialize_auth_tools(self, auth_tools: Optional[Auth]) -> Optional[dict]:
"""Serialize auth_tools to dictionary."""
if auth_tools is None:
return None
return AuthSerializer().to_dict(auth_tools)

@field_validator('auth_tools', mode='before')
@classmethod
def validate_auth_tools(cls, v: Any) -> Optional[Auth]:
"""Validate and deserialize auth_tools from dictionary."""
if v is None:
return None
if isinstance(v, Auth):
return v
if isinstance(v, dict):
return AuthSerializer().validate_dict(v)
raise ValueError(f"auth_tools must be None, Auth instance, or dict, got {type(v)}")


class FileCallTemplateSerializer(Serializer[FileCallTemplate]):
"""REQUIRED
Serializer for FileCallTemplate."""

def to_dict(self, obj: FileCallTemplate) -> dict:
"""REQUIRED
Convert a FileCallTemplate to a dictionary."""
return obj.model_dump()

def validate_dict(self, obj: dict) -> FileCallTemplate:
"""REQUIRED
Validate and convert a dictionary to a FileCallTemplate."""
try:
return FileCallTemplate.model_validate(obj)
except Exception as e:
raise UtcpSerializerValidationError("Invalid FileCallTemplate: " + traceback.format_exc())
Loading