server.py•9.56 kB
"""MCP CopyQ Server - Main entry point."""
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .tools.read import mcp_read
from .tools.write import mcp_write
from .tools.validate import mcp_validate
# Create server instance
server = Server("mcp-copyq")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="mcp_read",
description="""Read from CopyQ clipboard manager.
MCP tabs (full read/write access):
- mcp/info - general information storage
- mcp/заметки - notes storage
- mcp/workspace - projects (supports subtabs like workspace/myproject)
External tabs (READ-ONLY access with scope="all" or scope="external"):
- All other CopyQ tabs like "&clipboard", personal tabs, etc.
- Use scope="all" to see all tabs, scope="external" for non-mcp only
Modes:
- tree: Get tab structure with previews. Use FIRST to see available tabs.
- list: Get items from tab with pagination
- item: Get single item with full content (text, tags, note)
- search: Search by regex across tabs
Parameters:
- mode (required): "tree" | "list" | "item" | "search"
- tab: For mcp tabs use relative path "info", "workspace/proj1". For external use full name "&clipboard"
- scope: "mcp" (default) | "all" | "external" - controls which tabs are accessible
- index: Item index (for mode=item)
- query: Search regex (for mode=search)
Examples:
- tree of mcp tabs: mode="tree"
- tree of ALL tabs: mode="tree", scope="all"
- read external tab: mode="list", tab="&clipboard", scope="external"
- search everywhere: mode="search", query="pattern", scope="all"
Errors: TAB_NOT_FOUND, INDEX_OUT_OF_BOUNDS, INVALID_MODE""",
inputSchema={
"type": "object",
"properties": {
"mode": {
"type": "string",
"enum": ["tree", "list", "item", "search"],
"description": "Operation mode"
},
"tab": {
"type": "string",
"description": "Tab path: 'info', 'workspace/proj1' or full external name '&clipboard'"
},
"scope": {
"type": "string",
"enum": ["mcp", "all", "external"],
"default": "mcp",
"description": "Tab scope: mcp (default, read/write), all (read external), external (only non-mcp, read-only)"
},
"index": {
"type": "integer",
"description": "Item index (for mode=item)"
},
"query": {
"type": "string",
"description": "Search regex (for mode=search)"
},
"search_in": {
"type": "string",
"enum": ["text", "note", "tags", "all"],
"default": "all"
},
"max_depth": {
"type": "integer",
"minimum": 1,
"maximum": 10,
"default": 2
},
"max_items": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": 20
},
"skip": {
"type": "integer",
"minimum": 0,
"default": 0
},
"include_text": {
"type": "boolean",
"default": True
},
"include_tags": {
"type": "boolean",
"default": True
},
"include_note": {
"type": "boolean",
"default": False
}
},
"required": ["mode"]
}
),
Tool(
name="mcp_write",
description="""Write to CopyQ clipboard manager.
IMPORTANT: All data stored under "mcp/" prefix. When you use tab="info", actual CopyQ path is "mcp/info".
Available tabs:
- info - general storage (use tab="info")
- заметки - notes (use tab="заметки")
- workspace - projects, supports subtabs (use tab="workspace" or "workspace/myproject")
Modes:
- add: Add item to tab. Params: tab, text, tags (optional), note (optional)
- update: Update item. Params: tab, index, field, text/tags/note
- delete: Delete item. Params: tab, index
- move: Move item. Params: tab, index, to_tab
- tab_create: Create subtab in workspace only. Params: path (e.g. "workspace/newproject")
- tab_delete: Delete subtab. Params: path
Response shows full_path (e.g. "mcp/info") confirming where data was written.
Errors: TAB_NOT_FOUND, INDEX_OUT_OF_BOUNDS, PERMISSION_DENIED, MISSING_PARAM""",
inputSchema={
"type": "object",
"properties": {
"mode": {
"type": "string",
"enum": ["add", "update", "delete", "move", "tab_create", "tab_delete"],
"description": "Operation mode"
},
"tab": {
"type": "string",
"description": "Tab path"
},
"index": {
"type": "integer",
"description": "Item index"
},
"text": {
"type": "string",
"description": "Text content"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Tags list"
},
"note": {
"type": "string",
"description": "Note content"
},
"field": {
"type": "string",
"enum": ["text", "note", "tags"],
"description": "Field to update"
},
"edit_mode": {
"type": "string",
"enum": ["replace", "append", "prepend", "substitute", "remove"],
"default": "replace"
},
"match": {
"type": "string",
"description": "String to match (for substitute)"
},
"to_tab": {
"type": "string",
"description": "Destination tab (for move)"
},
"path": {
"type": "string",
"description": "Tab path (for tab_create/tab_delete)"
},
"intent": {
"type": "string",
"enum": ["execute", "preview"],
"default": "execute"
}
},
"required": ["mode"]
}
),
Tool(
name="mcp_validate",
description="""Validate parameters for mcp_read or mcp_write without executing.
Use to check if a call will succeed before making it.
Parameters:
- tool (required): "read" | "write"
- params (required): Parameters object to validate
Returns:
- valid: true/false
- errors: List of error codes and messages
- warnings: List of warnings
- estimated_tokens: Estimated response token count""",
inputSchema={
"type": "object",
"properties": {
"tool": {
"type": "string",
"enum": ["read", "write"],
"description": "Tool to validate"
},
"params": {
"type": "object",
"description": "Parameters to validate"
}
},
"required": ["tool", "params"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "mcp_read":
result = await mcp_read(**arguments)
elif name == "mcp_write":
result = await mcp_write(**arguments)
elif name == "mcp_validate":
result = await mcp_validate(**arguments)
else:
result = f"error|UNKNOWN_TOOL|Tool '{name}' not found"
return [TextContent(type="text", text=result)]
except Exception as e:
return [TextContent(type="text", text=f"error|INTERNAL|{str(e)}")]
async def run():
"""Run the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def main():
"""Entry point."""
asyncio.run(run())
if __name__ == "__main__":
main()