write_kv_value
Store key-value data in Cloudflare Workers KV storage to persist application data, cache content, or manage configuration settings with optional expiration and metadata.
Instructions
Write a key-value pair to Workers KV storage. Can store text or metadata.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| account_id | No | Account ID (uses default from config if not provided) | |
| namespace_id | Yes | The KV namespace ID | |
| key | Yes | The key to write | |
| value | Yes | The value to store | |
| expiration_ttl | No | Number of seconds for the key to expire | |
| metadata | No | Arbitrary JSON metadata to store with the key |
Input Schema (JSON Schema)
{
"properties": {
"account_id": {
"description": "Account ID (uses default from config if not provided)",
"type": "string"
},
"expiration_ttl": {
"description": "Number of seconds for the key to expire",
"type": "number"
},
"key": {
"description": "The key to write",
"type": "string"
},
"metadata": {
"description": "Arbitrary JSON metadata to store with the key",
"type": "object"
},
"namespace_id": {
"description": "The KV namespace ID",
"type": "string"
},
"value": {
"description": "The value to store",
"type": "string"
}
},
"required": [
"namespace_id",
"key",
"value"
],
"type": "object"
}
Implementation Reference
- The core handler function that executes the write_kv_value tool logic. It constructs the Cloudflare KV API URL and performs a PUT request with the value, optional expiration_ttl, and metadata.async def _write_kv_value(self, args: dict) -> str: """Write KV value.""" account_id = args.get("account_id") or self.account_id if not account_id: raise ValueError("Account ID is required. Provide it in args or config.") url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/storage/kv/namespaces/{args['namespace_id']}/values/{args['key']}" params = {} if args.get("expiration_ttl"): params["expiration_ttl"] = args["expiration_ttl"] if args.get("metadata"): params["metadata"] = json.dumps(args["metadata"]) headers = { "Authorization": f"Bearer {self.api_token}", "Content-Type": "text/plain", } response = await self.client.put( url, content=args["value"], params=params, headers=headers ) response.raise_for_status() return "KV value written successfully"
- src/cloudflare_mcp_server/__init__.py:304-331 (registration)Registers the write_kv_value tool with the MCP server in the list_tools handler, including the tool name, description, and detailed input schema.Tool( name="write_kv_value", description="Write a key-value pair to Workers KV storage. Can store text or metadata.", inputSchema={ "type": "object", "properties": { "account_id": { "type": "string", "description": "Account ID (uses default from config if not provided)", }, "namespace_id": { "type": "string", "description": "The KV namespace ID", }, "key": {"type": "string", "description": "The key to write"}, "value": {"type": "string", "description": "The value to store"}, "expiration_ttl": { "type": "number", "description": "Number of seconds for the key to expire", }, "metadata": { "type": "object", "description": "Arbitrary JSON metadata to store with the key", }, }, "required": ["namespace_id", "key", "value"], }, ),
- Defines the input schema for the write_kv_value tool, specifying parameters like namespace_id, key, value (required), and optional account_id, expiration_ttl, metadata.inputSchema={ "type": "object", "properties": { "account_id": { "type": "string", "description": "Account ID (uses default from config if not provided)", }, "namespace_id": { "type": "string", "description": "The KV namespace ID", }, "key": {"type": "string", "description": "The key to write"}, "value": {"type": "string", "description": "The value to store"}, "expiration_ttl": { "type": "number", "description": "Number of seconds for the key to expire", }, "metadata": { "type": "object", "description": "Arbitrary JSON metadata to store with the key", }, }, "required": ["namespace_id", "key", "value"], },