Spaces:
Running
Running
| """Actual MCP stdio fixture. Only an owned disposable root and fixed tools.""" | |
| import argparse,asyncio,hashlib,hmac,json,os,subprocess,sys | |
| from pathlib import Path,PurePosixPath | |
| from mcp.server.fastmcp import FastMCP,Context | |
| parser=argparse.ArgumentParser();parser.add_argument('--root',required=True);args=parser.parse_args() | |
| ROOT=Path(args.root).resolve();TOKEN=os.environ['NEEDLE_MCP_TEST_TOKEN'] | |
| server=FastMCP('needle-disposable-fixture',log_level='ERROR') | |
| def check(ctx,name): | |
| meta=ctx.request_context.meta | |
| provided=meta.model_dump().get('capability','') if meta else '' | |
| if not isinstance(provided,str) or not hmac.compare_digest(provided,TOKEN):raise ValueError('Unauthorized MCP fixture request') | |
| policy=json.loads((ROOT/'.git'/'needle-executor-policy.json').read_text()) | |
| if name not in policy['allowed_tools']:raise ValueError('Current executor permission denied') | |
| def target(path): | |
| try:relative=PurePosixPath(path).relative_to('/project') | |
| except ValueError:raise ValueError('Target outside disposable workspace') | |
| resolved=(ROOT/relative).resolve() | |
| if '..' in relative.parts or not resolved.is_relative_to(ROOT) or relative.as_posix() not in {'app.py','test_app.py'}:raise ValueError('Target not authorized') | |
| return resolved | |
| def command(*args): | |
| p=subprocess.run(args,cwd=ROOT,text=True,capture_output=True,timeout=15,env=dict(os.environ,PYTHONDONTWRITEBYTECODE='1')) | |
| return {'returncode':p.returncode,'stdout':p.stdout.replace(str(ROOT),'/project'),'stderr':p.stderr.replace(str(ROOT),'/project')} | |
| def list_files(ctx:Context)->dict: | |
| """List repository files.""" | |
| check(ctx,'list_files');return {'files':['app.py','test_app.py']} | |
| def search_files(query:str,ctx:Context)->dict: | |
| """Search source text.""" | |
| check(ctx,'search_files');return {'matches':[f'app.py:{i}:{line}' for i,line in enumerate((ROOT/'app.py').read_text().splitlines(),1) if query in line]} | |
| def read_file(path:str,ctx:Context)->dict: | |
| """Read a source file.""" | |
| check(ctx,'read_file');return {'content':target(path).read_text()} | |
| def write_file(path:str,content:str,ctx:Context)->dict: | |
| """Write exact file contents.""" | |
| check(ctx,'write_file');data=content.encode('utf-8');target(path).write_bytes(data) | |
| return {'bytes':len(data),'sha256':hashlib.sha256(data).hexdigest()} | |
| def run_tests(ctx:Context)->dict: | |
| """Execute the test suite.""" | |
| check(ctx,'run_tests');return command(sys.executable,'-m','pytest','-qq','--tb=line','--no-summary','-p','no:cacheprovider') | |
| def git_status(ctx:Context)->dict: | |
| """Inspect repository status.""" | |
| check(ctx,'git_status');return command('git','status','--short','--untracked-files=no') | |
| def git_diff(ctx:Context)->dict: | |
| """Inspect source changes.""" | |
| check(ctx,'git_diff');return command('git','diff','--','app.py') | |
| def git_commit(message:str,ctx:Context)->dict: | |
| """Commit changes with an explicit message.""" | |
| check(ctx,'git_commit');command('git','add','app.py');return command('git','commit','-qm',message) | |
| async def probe_delay(ctx:Context)->dict: | |
| """Test-only cancellation probe, never in the gateway catalogue.""" | |
| check(ctx,'probe_delay');await asyncio.sleep(2);return {'done':True} | |
| def probe_error(ctx:Context)->dict: | |
| """Test-only fixed error probe, never in the gateway catalogue.""" | |
| check(ctx,'probe_error');raise ValueError('Deliberate harmless fixture error') | |
| if __name__=='__main__':server.run(transport='stdio') | |