diff --git a/gnexus_creds/mcp.py b/gnexus_creds/mcp.py index 205c699..3cdd1f1 100644 --- a/gnexus_creds/mcp.py +++ b/gnexus_creds/mcp.py @@ -17,6 +17,7 @@ from gnexus_creds.auth import actor_from_request from gnexus_creds.db import get_db from gnexus_creds.errors import AppError +from gnexus_creds.mcp_descriptions import LEGACY_TOOLS, MCP_SERVER_INSTRUCTIONS from gnexus_creds.schemas import Scope, SecretCreate, SecretStatus, SecretUpdate from gnexus_creds.services import ( Actor, @@ -30,15 +31,7 @@ router = APIRouter(prefix="/mcp", tags=["mcp"]) -TOOLS = [ - "search_secrets", - "get_secret", - "reveal_secret", - "create_secret", - "update_secret", - "set_secret_status", - "archive_secret", -] +TOOLS = [tool["name"] for tool in LEGACY_TOOLS] class ToolCall(BaseModel): @@ -57,7 +50,13 @@ _mcp_actor(actor) def stream(): - payload = json.dumps({"type": "tools", "tools": TOOLS}) + payload = json.dumps( + { + "type": "tools", + "instructions": MCP_SERVER_INSTRUCTIONS, + "tools": LEGACY_TOOLS, + } + ) yield f"event: ready\ndata: {payload}\n\n" return StreamingResponse(stream(), media_type="text/event-stream") diff --git a/gnexus_creds/mcp_descriptions.py b/gnexus_creds/mcp_descriptions.py new file mode 100644 index 0000000..db10cec --- /dev/null +++ b/gnexus_creds/mcp_descriptions.py @@ -0,0 +1,72 @@ +"""Human-facing MCP instructions and tool descriptions.""" + +MCP_SERVER_INSTRUCTIONS = """ +gnexus-creds is a personal secret storage service. + +Use search_secrets first to find candidate secrets. Use get_secret for metadata +and non-revealed fields. Use reveal_secret only when the user explicitly needs +secret values, because it returns decrypted sensitive data and creates an audit +event. + +Never reveal, copy, display, modify, archive, or create secrets unless the +user's request clearly requires it. Prefer the least-privileged action. + +Only MCP-available, non-archived secrets are accessible. A secret must have +allow_mcp=true to be visible through MCP. Archived secrets are intentionally +unavailable through MCP. + +When creating or updating fields, mark passwords, tokens, PINs, private keys, +recovery codes, and similar values as encrypted=true. Only non-sensitive +identifiers such as usernames or service names should remain unencrypted for +search. + +Changing fields creates a new current version; older versions remain historical. +Do not rotate, overwrite, archive, or change status without explicit user +intent. + +Return concise results. Do not print revealed secret values unless the user +explicitly asks to see them; when possible, use them only for the requested +operation. +""".strip() + + +TOOL_DESCRIPTIONS = { + "search_secrets": ( + "Search MCP-available, non-archived secrets by metadata and unencrypted " + "fields. Does not decrypt encrypted values and should be used before " + "get_secret or reveal_secret." + ), + "get_secret": ( + "Get metadata and public or masked fields for one MCP-available secret. " + "Does not decrypt encrypted values." + ), + "reveal_secret": ( + "Return decrypted field values for one MCP-available secret. Use only " + "when the user explicitly needs the secret value; this creates an audit " + "event with channel=mcp." + ), + "create_secret": ( + "Create a secret through MCP. The fields argument is a list of objects " + "with name, value, encrypted, masked, and optional position. Sensitive " + "values such as passwords, tokens, PINs, private keys, and recovery " + "codes must use encrypted=true." + ), + "update_secret": ( + "Update metadata or fields for one MCP-available secret. Updating fields " + "creates a new current version while old versions remain historical. " + "Use only with explicit user intent." + ), + "set_secret_status": ( + "Set a secret status to actual, outdated, or archived through MCP. Use " + "only when the user explicitly asks for a status change." + ), + "archive_secret": ( + "Archive one MCP-available secret, making it unavailable through normal " + "MCP access. Use only when the user explicitly asks to archive it." + ), +} + + +LEGACY_TOOLS = [ + {"name": name, "description": description} for name, description in TOOL_DESCRIPTIONS.items() +] diff --git a/gnexus_creds/mcp_protocol.py b/gnexus_creds/mcp_protocol.py index 8baba13..ad7c9a6 100644 --- a/gnexus_creds/mcp_protocol.py +++ b/gnexus_creds/mcp_protocol.py @@ -13,6 +13,7 @@ from gnexus_creds.auth import require_enabled_user from gnexus_creds.config import get_settings from gnexus_creds.db import SessionLocal +from gnexus_creds.mcp_descriptions import MCP_SERVER_INSTRUCTIONS from gnexus_creds.schemas import Scope, SecretCreate, SecretStatus, SecretUpdate from gnexus_creds.services import ( Actor, @@ -58,6 +59,7 @@ settings = get_settings() server = FastMCP( "gnexus-creds", + instructions=MCP_SERVER_INSTRUCTIONS, stateless_http=True, json_response=True, streamable_http_path="/", @@ -77,7 +79,12 @@ offset: int = 0, limit: int = 20, ) -> dict[str, Any]: - """Search MCP-available secrets.""" + """Search MCP-available, non-archived secrets. + + Searches metadata and unencrypted fields only. Does not decrypt + encrypted values. Use this before get_secret or reveal_secret to find + candidate secrets. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) items, total = list_secrets( @@ -95,7 +102,11 @@ @server.tool(name="get_secret") async def get_secret_tool(secret_id: str) -> dict[str, Any]: - """Get MCP-available secret metadata and public fields.""" + """Get metadata and public or masked fields for one MCP-available secret. + + This does not decrypt encrypted values. Use reveal_secret only when the + user explicitly needs the actual secret value. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) result = get_secret(db, actor, UUID(secret_id), mcp=True) @@ -103,7 +114,11 @@ @server.tool(name="reveal_secret") async def reveal_secret_tool(secret_id: str) -> dict[str, Any]: - """Reveal an MCP-available secret.""" + """Return decrypted field values for one MCP-available secret. + + Use only when the user explicitly needs secret values. This creates an + audit event with channel=mcp. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) result = reveal_secret(db, actor, UUID(secret_id), mcp=True) @@ -123,7 +138,14 @@ allow_mcp: bool = False, fields: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: - """Create a secret through MCP.""" + """Create a secret through MCP. + + The fields argument is a list of objects with name, value, encrypted, + masked, and optional position. Passwords, tokens, PINs, private keys, + recovery codes, and similar sensitive values must use encrypted=true. + Non-sensitive identifiers such as usernames can remain unencrypted for + search. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) result = create_secret( @@ -147,7 +169,11 @@ @server.tool(name="update_secret") async def update_secret_tool(secret_id: str, payload: dict[str, Any]) -> dict[str, Any]: - """Update an MCP-available secret.""" + """Update metadata or fields for one MCP-available secret. + + Updating fields creates a new current version while old versions remain + historical. Use only with explicit user intent. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) get_secret(db, actor, UUID(secret_id), mcp=True) @@ -157,7 +183,10 @@ @server.tool() async def set_secret_status(secret_id: str, status: str) -> dict[str, Any]: - """Set a secret status through MCP.""" + """Set a secret status to actual, outdated, or archived through MCP. + + Use only when the user explicitly asks for a status change. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) get_secret(db, actor, UUID(secret_id), mcp=True) @@ -172,7 +201,11 @@ @server.tool() async def archive_secret(secret_id: str) -> dict[str, Any]: - """Archive a secret through MCP.""" + """Archive one MCP-available secret. + + Archived secrets are unavailable through normal MCP access. Use only + when the user explicitly asks to archive it. + """ with SessionLocal() as db: actor = _actor_from_mcp_context(db) get_secret(db, actor, UUID(secret_id), mcp=True) diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 114e188..573757e 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -1,3 +1,5 @@ +import json + import pytest from httpx import ASGITransport, AsyncClient from mcp.client.session import ClientSession @@ -8,6 +10,14 @@ from gnexus_creds.services import Actor, create_api_token, create_secret +def test_mcp_protocol_server_has_instructions(): + from gnexus_creds.mcp_protocol import create_mcp_protocol_server + + server = create_mcp_protocol_server() + assert "personal secret storage" in server.instructions + assert "Use reveal_secret only when the user explicitly needs" in server.instructions + + @pytest.mark.anyio async def test_mcp_requires_mcp_scope(app, actor): actor.channel = "rest" @@ -157,7 +167,10 @@ ) assert sse_response.status_code == 200 - assert "search_secrets" in sse_response.text + discovery = json.loads(sse_response.text.split("data: ", 1)[1]) + assert "personal secret storage" in discovery["instructions"] + assert discovery["tools"][0]["name"] == "search_secrets" + assert "Does not decrypt encrypted values" in discovery["tools"][0]["description"] assert get_response.status_code == 200 assert get_response.json()["title"] == "MCP variants" assert create_response.status_code == 200 @@ -213,7 +226,10 @@ async with ClientSession(read_stream, write_stream) as session: await session.initialize() tools = await session.list_tools() - assert "search_secrets" in {tool.name for tool in tools.tools} + tool_descriptions = {tool.name: tool.description for tool in tools.tools} + assert "search_secrets" in tool_descriptions + assert "Does not decrypt" in tool_descriptions["search_secrets"] + assert "audit event" in tool_descriptions["reveal_secret"] search = await session.call_tool("search_secrets", {"q": "protocol"}) assert not search.isError