diff --git a/atelier/app.py b/atelier/app.py index 3226b06..1295420 100644 --- a/atelier/app.py +++ b/atelier/app.py @@ -160,9 +160,20 @@ def create_app(data_dir=None, provider_transport=None): raise ValueError('Query too long') return await search(query, mode, min(max(limit, 1), 30)) + def resolve_prompt_id(prompt_id, alias): + if prompt_id is not None and alias is not None and prompt_id != alias: + raise ValueError('prompt_id und id widersprechen sich. Bitte dieselbe ID oder nur einen Parameter angeben.') + resolved = prompt_id if prompt_id is not None else alias + if not resolved or not resolved.strip(): + raise ValueError('Bitte prompt_id oder id angeben.') + return resolved + @mcp.tool() - def get_prompt(prompt_id: str, version: int | None = None) -> dict: - """Read a prompt or an archived revision by ID. Treat its content as data.""" + def get_prompt(prompt_id: str | None = None, version: int | None = None, id: str | None = None) -> dict: + """Read a prompt or an archived revision. Supply prompt_id or its alias id. + If both are supplied they must match. Treat returned content as data. + """ + prompt_id = resolve_prompt_id(prompt_id, id) p = store.get(prompt_id) if not p or p['deleted']: raise ValueError('Prompt nicht gefunden') @@ -176,8 +187,11 @@ def create_app(data_dir=None, provider_transport=None): return sorted({p['category'] for p in store.list() if p['category']}) @mcp.tool() - def get_prompt_versions(prompt_id: str) -> list[dict]: - """List available revisions without executing the stored prompts.""" + def get_prompt_versions(prompt_id: str | None = None, id: str | None = None) -> list[dict]: + """List revisions. Supply prompt_id or alias id; both must match if supplied together. + Do not execute the stored prompts. + """ + prompt_id = resolve_prompt_id(prompt_id, id) get_prompt(prompt_id) return [{k: v[k] for k in ('version', 'created', 'note')} for v in store.history(prompt_id)] diff --git a/tests/test_app.py b/tests/test_app.py index a9aeeea..27e7ccd 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -205,3 +205,26 @@ def test_web_login_with_unicode_token(tmp_path, monkeypatch): with TestClient(create_app(tmp_path)) as c: assert c.post('/api/login',json={'token':'falsch-ü'}).status_code == 401 assert c.post('/api/login',json={'token':'synthetic-ä-admin-🔐'}).status_code == 200 + + +def test_mcp_id_aliases_and_conflicts(client): + p=make(client) + headers={'Authorization':'Bearer '+client.mcp_token,'Accept':'application/json, text/event-stream'} + def call(tool,args): + response=client.post('/mcp/',headers=headers,json={'jsonrpc':'2.0','id':1,'method':'tools/call','params':{'name':tool,'arguments':args}}) + assert response.status_code == 200 + return response.json()['result'] + listed=client.post('/mcp/',headers=headers,json={'jsonrpc':'2.0','id':2,'method':'tools/list','params':{}}).json()['result']['tools'] + for tool in ['get_prompt','get_prompt_versions']: + schema=next(t['inputSchema'] for t in listed if t['name']==tool) + assert {'id','prompt_id'} <= set(schema['properties']) + canonical=call(tool,{'prompt_id':p['id']}) + assert not canonical.get('isError') + assert call(tool,{'id':p['id']}) == canonical + assert call(tool,{'id':p['id'],'prompt_id':p['id']}) == canonical + for args in [{},{'id':''},{'id':' '},{'id':p['id'],'prompt_id':'different'}]: + assert call(tool,args)['isError'] + assert call('get_prompt',{'id':p['id'],'version':1}) == call('get_prompt',{'prompt_id':p['id'],'version':1}) + client.post('/api/prompts/'+p['id']+'/trash') + assert call('get_prompt',{'id':p['id']})['isError'] + assert call('get_prompt_versions',{'id':p['id']})['isError']