mcp_generator.cpython-311.pyc•17.9 kB
�
�X�g]3 � � � d dl mZmZ d dlmZ d dlZd dlZddlmZm Z m
Z
ddlmZ d dl
Z
d dlZd dlZd dlmZmZ d dlmZ d d l mZmZmZ G d
� d� � ZdS )� )�Dict�List)�PathN� )�FunctionDefinition�FunctionParameter�FunctionType)�LLMMethodGenerator)�FastAPI�
HTTPException)� BaseModel)r �Any�Optionalc � � e Zd Zdedededefd�Zd� Zd� Zd� Z d � Z
d
efd�Zd� Z
d
edefd�Zd� Zdefd�Zdefd�Zdefd�ZdS )�MCPGenerator�analysis�
output_dir�
contract_name�openai_api_keyc �| � || _ || _ || _ t t |dz � � |�� � | _ dS )z7Initialize the MCP generator with ABI analysis results.�cache)� cache_dirr N)r r r r
�str�
llm_generator)�selfr r r r s �9/Users/arjun/repos/sparkmango/mcp_server/mcp_generator.py�__init__zMCPGenerator.__init__ sH � � ��
�$���*���/��*�w�.�/�/�)�
�
�
����� c � �� K � | � � � | � � � | � � � | � � � � d{V �� | � � � dS )z'Generate the MCP server implementation.N)�_create_directory_structure�_generate_state_variables�_generate_server_file�_generate_methods�_generate_documentation)r s r �generatezMCPGenerator.generate s� � � � �
�(�(�*�*�*�
�&�&�(�(�(�
�"�"�$�$�$� �$�$�&�&�&�&�&�&�&�&�&�
�$�$�&�&�&�&�&r c �r � | j | j dz | j dz | j dz | j dz | j dz g}|D ]}|� d�� � �|D ]/}|dz }|� � � s|� � � �0| j dz }|� � � s|� � � d S d S )
z<Create the necessary directory structure for the MCP server.�methods�state�tests�docsr T)�exist_ok�__init__.pyN)r �mkdir�exists�touch)r �directories� directory� init_file� root_inits r r z(MCPGenerator._create_directory_structure* s� � �
�O��O�i�'��O�g�%��O�g�%��O�f�$��O�g�%�
�� %� +� +�I��O�O�T�O�*�*�*�*� %� "� "�I�!�M�1�I��#�#�%�%�
"����!�!�!�� �O�m�3� ����!�!� ��O�O������ � r c �� � d� | j �� � }| j dz }t |d� � 5 }|� |� � ddd� � dS # 1 swxY w Y dS )z"Generate the main MCP server file.a@
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any, Optional
import json
import logging
import sys
import importlib.util
from pathlib import Path
# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Add the current directory to Python path
current_dir = Path(__file__).parent
sys.path.append(str(current_dir))
logger.debug("Added %s to Python path", current_dir)
try:
from methods import *
logger.debug("Successfully imported methods")
except ImportError as e:
logger.error("Failed to import methods: %s", e)
raise
try:
from state import State
logger.debug("Successfully imported State")
except ImportError as e:
logger.error("Failed to import State: %s", e)
raise
app = FastAPI(
title="{contract_name} MCP Server",
description="Model Context Protocol server for {contract_name} smart contract",
version="1.0.0"
)
class MCPRequest(BaseModel):
"""Base model for MCP requests."""
method: str
params: Dict[str, Any]
context: Optional[Dict[str, Any]] = None
class MCPResponse(BaseModel):
"""Base model for MCP responses."""
result: Any
context: Optional[Dict[str, Any]] = None
# Initialize contract state
try:
state = State()
logger.debug("Successfully initialized State")
except Exception as e:
logger.error("Failed to initialize State: %s", e)
raise
def load_method(method_name: str):
"""Dynamically load a method implementation."""
logger.debug("Loading method %s", method_name)
method_path = Path(__file__).parent / 'methods' / f'{{method_name}}.py'
if not method_path.exists():
logger.error("Method file not found: %s", method_path)
raise HTTPException(status_code=404, detail=f"Method {{method_name}} not found")
try:
spec = importlib.util.spec_from_file_location(method_name, str(method_path))
module = importlib.util.module_from_spec(spec)
sys.modules[method_name] = module
spec.loader.exec_module(module)
method = getattr(module, method_name)
logger.debug("Successfully loaded method %s", method_name)
return method
except Exception as e:
logger.error("Failed to load method %s: %s", method_name, e)
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp", response_model=MCPResponse)
async def process_mcp_request(request: MCPRequest):
"""
Process an MCP request for the {contract_name} contract.
This endpoint accepts requests in the Model Context Protocol format and
routes them to the appropriate contract method implementation.
"""
logger.debug("Processing MCP request: %s", request.method)
try:
# Load and execute the method
method = load_method(request.method)
result = await method(state, **request.params)
logger.debug("Method %s executed successfully", request.method)
return MCPResponse(
result=result,
context=request.context
)
except HTTPException:
raise
except Exception as e:
logger.error("Error processing MCP request: %s", e)
raise HTTPException(status_code=500, detail=str(e))
if __name__ == '__main__':
import uvicorn
logger.info("Starting MCP server")
uvicorn.run(app, host="0.0.0.0", port=8000)
)r z server.py�wN)�formatr r �open�write)r �template�server_path�fs r r"