scripts/workflow_mcp_server.py

"""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')}

@server.tool()
def list_files(ctx:Context)->dict:
    """List repository files."""
    check(ctx,'list_files');return {'files':['app.py','test_app.py']}
@server.tool()
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]}
@server.tool()
def read_file(path:str,ctx:Context)->dict:
    """Read a source file."""
    check(ctx,'read_file');return {'content':target(path).read_text()}
@server.tool()
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()}
@server.tool()
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')
@server.tool()
def git_status(ctx:Context)->dict:
    """Inspect repository status."""
    check(ctx,'git_status');return command('git','status','--short','--untracked-files=no')
@server.tool()
def git_diff(ctx:Context)->dict:
    """Inspect source changes."""
    check(ctx,'git_diff');return command('git','diff','--','app.py')
@server.tool()
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)
@server.tool()
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}
@server.tool()
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')