如何构建 MCP 会议记录到行动管道

from mcp.server import Server
import json
from openai import OpenAI

server = Server('meeting-actions')
llm = OpenAI()

@server.tool()
async def extract_actions(meeting_notes: str) -> str:
    """Extract action items from meeting notes."""
    response = llm.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'system', 'content': '''Extract action items from meeting notes.
Return JSON array with: task, assignee, deadline, search_query (a web search
that would provide useful context for this task).'''},
            {'role': 'user', 'content': meeting_notes}],
        response_format={'type': 'json_object'},
        max_tokens=1000
    )
    return response.choices[0].message.content

import requests, os

API_KEY = os.environ['SCAVIO_API_KEY']

@server.tool()
async def enrich_action(search_query: str, task_context: str = '') -> str:
    """Enrich an action item with web search context."""
    resp = requests.post('https://api.scavio.dev/api/v1/search',
        headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
        json={'query': search_query, 'country_code': 'us'})
    results = resp.json().get('organic_results', [])[:5]
    enrichment = {
        'query': search_query,
        'key_findings': [
            {'source': r['link'], 'title': r['title'],
             'insight': r.get('snippet', '')}
            for r in results
        ],
        'sources_count': len(results)
    }
    return json.dumps(enrichment, indent=2)

@server.tool()
async def meeting_to_actions(meeting_notes: str) -> str:
    """Extract action items from meeting notes and enrich each with search context."""
    # Step 1: Extract actions
    actions_json = await extract_actions(meeting_notes)
    actions = json.loads(actions_json).get('actions', [])
    # Step 2: Enrich each action
    enriched = []
    for action in actions:
        search_query = action.get('search_query', action.get('task', ''))
        enrichment_json = await enrich_action(search_query, action.get('task', ''))
        enrichment = json.loads(enrichment_json)
        enriched.append({
            **action,
            'enrichment': enrichment
        })
    result = {
        'total_actions': len(enriched),
        'credits_used': len(enriched),  # 1 search per action
        'cost': f'${len(enriched) * 0.005:.3f}',
        'actions': enriched
    }
    return json.dumps(result, indent=2)

import asyncio
from mcp.server.stdio import stdio_server

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == '__main__':
    asyncio.run(main())

# .mcp.json configuration:
# {
#   "mcpServers": {
#     "meeting-actions": {
#       "command": "python",
#       "args": ["meeting_actions_server.py"],
#       "env": {
#         "SCAVIO_API_KEY": "your_key",
#         "OPENAI_API_KEY": "your_key"
#       }
#     }
#   }
# }

# Test the pipeline directly (without MCP client):
import asyncio

async def test():
    notes = """Product sync 2026-05-13:
- Sarah: Research competitor pricing for CRM tools, need comparison by Friday
- Mike: Look into TikTok API options for our influencer vetting feature
- Lisa: Find best practices for AI Overview optimization, present next week
- Team: Evaluate n8n vs Zapier for our lead gen automation"""
    result = await meeting_to_actions(notes)
    data = json.loads(result)
    print(f'Extracted {data["total_actions"]} actions (cost: {data["cost"]})')
    for a in data['actions']:
        print(f'\n  Task: {a["task"]}')
        print(f'  Assignee: {a.get("assignee", "unassigned")}')
        print(f'  Context: {len(a["enrichment"]["key_findings"])} sources found')

asyncio.run(test())