【2026 最新】MCP Tasks 教學：Claude Code 補問、無人值守與取消流程六步實作

import json, sqlite3, sys, time, uuid
from datetime import datetime, timezone

VERSION = '2026-07-28'
EXT = 'io.modelcontextprotocol/tasks'
META = 'io.modelcontextprotocol/'
db = sqlite3.connect(sys.argv[1] if len(sys.argv) > 1 else 'mock.db')
db.execute('CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, body TEXT)')
TERMINAL = {'completed', 'failed', 'cancelled'}

class CapabilityError(Exception):
    def __init__(self, required):
        self.required = required

def require_form(caps):
    elicit = caps.get('elicitation')
    if not isinstance(elicit, dict) or (elicit and 'form' not in elicit):
        raise CapabilityError({'elicitation':{'form':{}}})

def stamp():
    return datetime.now(timezone.utc).isoformat()

def save(t):
    db.execute('INSERT OR REPLACE INTO jobs VALUES (?,?)',
               (t['taskId'], json.dumps(t)))
    db.commit()

def status(t, value):
    t['status'], t['lastUpdatedAt'] = value, stamp()
    if value != 'input_required':
        t.pop('inputRequests', None)

def load(key):
    row = db.execute('SELECT body FROM jobs WHERE id=?', (key,)).fetchone()
    if not row:
        raise ValueError('Unknown task')
    t = json.loads(row[0])
    if t['status'] not in TERMINAL and time.time() >= t['_deadline']:
        status(t, 'cancelled')
        t['statusMessage'] = 'Demo deadline reached; no report generated'
        save(t)
    return t

def public(t, kind='complete'):
    return dict(resultType=kind, **{k:v for k,v in t.items() if not k.startswith('_')})

def dispatch(method, p):
    meta = p.get('_meta', {})
    if meta.get(META+'protocolVersion') != VERSION:
        raise ValueError('Use protocol 2026-07-28')
    caps = meta.get(META+'clientCapabilities')
    if not isinstance(caps, dict):
        raise ValueError('clientCapabilities must be an object')
    if method.startswith('tasks/') and EXT not in caps.get('extensions', {}):
        raise CapabilityError({'extensions':{EXT:{}}})
    if method == 'server/discover':
        return {'resultType':'complete', 'supportedVersions':[VERSION],
                'capabilities':{'tools':{}, 'extensions':{EXT:{}}}}
    if method == 'tools/list':
        return {'resultType':'complete', 'tools':[{
            'name':'mock_report', 'description':'Fake report; asks for a title',
            'inputSchema':{'type':'object','properties':{},'additionalProperties':False}}]}
    if method == 'tools/call':
        if p.get('name') != 'mock_report':
            raise ValueError('Unknown tool')
        if EXT not in caps.get('extensions', {}):
            raise CapabilityError({'extensions':{EXT:{}}})
        require_form(caps)
        now = stamp()
        t = dict(taskId=uuid.uuid4().hex, status='working', createdAt=now,
                 lastUpdatedAt=now, ttlMs=None, pollIntervalMs=500,
                 _deadline=time.time()+30)
        save(t)
        return public(t, 'task')
    if method not in {'tasks/get','tasks/update','tasks/cancel'}:
        raise ValueError('Unknown method')
    t = load(p['taskId'])
    if method == 'tasks/get':
        if t['status'] in {'working', 'input_required'}:
            require_form(caps)
        if t['status'] == 'working':
            status(t, 'input_required')
            t['inputRequests'] = {'title-1':{'method':'elicitation/create','params':{
                'mode':'form','message':'Choose a fake report title',
                'requestedSchema':{'type':'object','properties':{
                    'title':{'type':'string','maxLength':40}},'required':['title']}}}}
            save(t)
        return public(t)
    if method == 'tasks/cancel':
        if t['status'] not in TERMINAL:
            status(t, 'cancelled')
            t['statusMessage'] = 'Cancelled; no report generated'
            save(t)
        return {'resultType':'complete'}
    answer = p.get('inputResponses', {}).get('title-1')
    if t['status'] == 'input_required' and answer is not None:
        # InputResponses values are direct elicitation results.
        value = answer
        if value.get('action') != 'accept':
            status(t, 'cancelled')
        else:
            title = value.get('content', {}).get('title')
            if not isinstance(title, str) or not 1 <= len(title) <= 40:
                raise ValueError('Title must contain 1..40 characters')
            status(t, 'completed')
            t['result'] = {'resultType':'complete', 'content':[
                {'type':'text','text':'FAKE REPORT: '+title}]}
        save(t)
    return {'resultType':'complete'}

for line in sys.stdin:
    try:
        req = json.loads(line)
        if 'id' not in req:
            continue
        reply = {'jsonrpc':'2.0','id':req['id']}
        try:
            reply['result'] = dispatch(req['method'], req.get('params', {}))
        except CapabilityError as e:
            reply['error'] = {'code':-32021,'message':'Missing required client capability',
                              'data':{'requiredCapabilities':e.required}}
        except (ValueError, KeyError, TypeError) as e:
            reply['error'] = {'code':-32602,'message':str(e)}
        print(json.dumps(reply), flush=True)
    except json.JSONDecodeError:
        print(json.dumps({'jsonrpc':'2.0','id':None,'error':{
            'code':-32700,'message':'Parse error'}}), flush=True)

import json, subprocess, sys, time
from pathlib import Path
root = Path(__file__).resolve().parent
p = subprocess.Popen([sys.executable, str(root/'mock.py'), str(root/'verify.db')],
                     stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
meta = {'io.modelcontextprotocol/protocolVersion':'2026-07-28',
        'io.modelcontextprotocol/clientCapabilities':{
            'extensions':{'io.modelcontextprotocol/tasks':{}}, 'elicitation':{'form':{}}}}
trace=[]
def call(method, expect_error=False, **params):
    params['_meta']=meta
    req={'jsonrpc':'2.0','id':len(trace)+1,'method':method,'params':params}
    p.stdin.write(json.dumps(req)+'\n');p.stdin.flush()
    reply=json.loads(p.stdout.readline());trace.append({'request':req,'response':reply})
    if expect_error:
        assert reply.get('error',{}).get('code') == -32021, reply
        return reply['error']
    assert 'error' not in reply, reply
    return reply['result']
def job():
    t=call('tools/call',name='mock_report',arguments={});assert t['status']=='working'
    t=call('tasks/get',taskId=t['taskId']);assert t['status']=='input_required'
    return t['taskId']
assert call('server/discover')['supportedVersions']==['2026-07-28']
a=job();call('tasks/update',taskId=a,inputResponses={'title-1':{'action':'accept','content':{'title':'Alpha'}}})
assert call('tasks/get',taskId=a)['status']=='completed'
# A late cancellation acknowledgement must not rewrite a terminal state.
call('tasks/cancel',taskId=a);assert call('tasks/get',taskId=a)['status']=='completed'
b=job();call('tasks/cancel',taskId=b)
call('tasks/update',taskId=b,inputResponses={'title-1':{'action':'accept','content':{'title':'Late'}}})
assert call('tasks/get',taskId=b)['status']=='cancelled'
c=job();call('tasks/update',taskId=c,inputResponses={'title-1':{'action':'decline'}})
assert call('tasks/get',taskId=c)['status']=='cancelled'
d=job();p.stdin.close();p.wait()
p = subprocess.Popen([sys.executable,str(root/'mock.py'),str(root/'verify.db')],stdin=subprocess.PIPE,stdout=subprocess.PIPE,text=True)
assert call('tasks/get',taskId=d)['status']=='input_required'
time.sleep(31)
assert call('tasks/get',taskId=d)['status']=='cancelled'
# Check each request's capability; earlier declaration must not carry over.
old=meta;meta={'io.modelcontextprotocol/protocolVersion':'2026-07-28','io.modelcontextprotocol/clientCapabilities':{}}
call('tools/call',expect_error=True,name='mock_report',arguments={})
call('tasks/get',expect_error=True,taskId=a)
meta = old
meta['io.modelcontextprotocol/clientCapabilities']['elicitation'] = {}
e = job()
meta['io.modelcontextprotocol/clientCapabilities'].pop('elicitation')
call('tasks/get',expect_error=True,taskId=e)
meta['io.modelcontextprotocol/clientCapabilities']['elicitation'] = {}
call('tasks/cancel',taskId=e)
p.stdin.close();p.wait()
(root/'trace.json').write_text(json.dumps(trace,ensure_ascii=False,indent=2))
print('PASS: accept, cancel, late cancel, late input, decline, restart lookup, deadline, missing capability')
print('Scope: local stdio mock only; not a Claude Code UI integration run')