Skip to main content
Glama
mcp_generator.cpython-311.pyc17.9 kB
� �X�g]3���ddlmZmZddlmZddlZddlZddlmZm Z m Z ddl m Z ddl Z ddlZddlZddlmZmZddlmZdd lmZmZmZGd �d ��ZdS) �)�Dict�List)�PathN�)�FunctionDefinition�FunctionParameter� FunctionType)�LLMMethodGenerator)�FastAPI� HTTPException)� BaseModel)r�Any�Optionalc��eZdZdedededefd�Zd�Zd�Zd�Z d �Z d e fd �Z d �Z d edefd�Zd�Zdefd�Zdefd�Zdefd�ZdS)� MCPGenerator�analysis� output_dir� contract_name�openai_api_keyc�|�||_||_||_tt |dz ��|���|_dS)z7Initialize the MCP generator with ABI analysis results.�cache)� cache_dirrN)rrrr �str� llm_generator)�selfrrrrs �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)rs r�generatezMCPGenerator.generates����� �(�(�*�*�*� �&�&�(�(�(� �"�"�$�$�$��$�$�&�&�&�&�&�&�&�&�&� �$�$�&�&�&�&�&rc�r�|j|jdz |jdz |jdz |jdz |jdz g}|D]}|�d����|D]/}|dz }|���s|����0|jdz }|���s|���d Sd S) z<Create the necessary directory structure for the MCP server.�methods�state�tests�docsrT)�exist_ok� __init__.pyN)r�mkdir�exists�touch)r� directories� directory� init_file� root_inits rr 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� � � � � � � rc���d�|j���}|jdz }t|d��5}|�|��ddd��dS#1swxYwYdS)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) )rz server.py�wN)�formatrr�open�write)r�template� server_path�fs rr"z"MCPGenerator._generate_server_fileCs���k�V �F��+�F�,�,�W �\�o� �3� � �+�s� #� #� �q� �G�G�H� � � � � � � � � � � � � � � ���� � � � � � s�A�A� Ac��`K�|jdD]}|�|���d{V���dS)z/Generate MCP implementations for each function.� functionsN)r�_generate_method_file)r�functions rr#zMCPGenerator._generate_methods�sO����� �k�2� 7� 7�H��,�,�X�6�6� 6� 6� 6� 6� 6� 6� 6� 6� 7� 7rr?c��K�|j�||jd���d{V��}t|jdz |j�d�z d��5}|�|��ddd��dS#1swxYwYdS)z5Generate an MCP implementation for a single function.�abiNr'z.pyr5)r�generate_methodrr7r�namer8)rr?�implementationr;s rr>z"MCPGenerator._generate_method_file�s����� $�1�A�A�(�D�M�Z_�L`�a�a�a�a�a�a�a�a���$�/�I�-�8�=�0E�0E�0E�E�s� K� K� $�q� �G�G�N� #� #� #� $� $� $� $� $� $� $� $� $� $� $� $���� $� $� $� $� $� $s�A5�5A9�<A9c�~�d}g}g}|jdD]]}|�|d��}|�|d�d|�d���|�d|d�d����^d �|��}d �|��}t |jd ���d d ���dd��}t |jdz dz d��5} | �|� |||�����ddd��dS#1swxYwYdS)z(Generate state variable implementations.aAfrom pydantic import BaseModel, Field from typing import Dict, Any, List, Optional import os import logging # Configure logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) class State(BaseModel): """State variables for the MCP contract.""" # Contract state contract_address: str abi: List[Dict[str, Any]] account: str # Contract variables {state_vars} def __init__(self, **data): """Initialize state variables.""" logger.debug("Initializing State with data: %s", data) # Set contract state from environment contract_address = os.getenv("CONTRACT_ADDRESS") logger.debug("Got contract_address from env: %s", contract_address) data["contract_address"] = contract_address abi = {abi} logger.debug("Setting ABI: %s", abi) data["abi"] = abi account = os.getenv("ACCOUNT_ADDRESS", "0x0000000000000000000000000000000000000000") logger.debug("Got account from env: %s", account) data["account"] = account # Initialize contract variables with defaults {init_vars} logger.debug("Calling super().__init__ with data: %s", data) super().__init__(**data) logger.debug("State initialization complete") �state_variables�typerCz : Optional[z] = Field(default=None)zdata['z '] = Nonez z rA�false�False�true�Truer(r,r5)� state_vars� init_varsrAN) r�_get_python_type�append�joinr�replacer7rr8r6) rr9rLrM�var� type_hint�state_vars_str� init_vars_str�abi_strr;s rr!z&MCPGenerator._generate_state_variables�s���*��X� �� ��=�!2�3� >� >�C��-�-�c�&�k�:�:�I� � � ��V��[�[��[�[�[� \� \� \� � � �<�c�&�k�<�<�<� =� =� =� =�!���z�2�2��$�)�)�)�4�4� ��d�m�E�*�+�+�3�3�G�W�E�E�M�M�f�V\�]�]�� �$�/�G�+�m�;�S� A� A� �Q� �G�G�H�O�O�)�'��$��� � � � � � � � � � � � � � � ���� � � � � � s�9,D2�2D6�9D6� solidity_type�returnc�@�ddddddd�}|�|d��S)z*Convert Solidity type to Python type hint.�int�boolr�bytes)�uint256�uint8r[�address�string�bytes32r)�get)rrW� type_mappings rrNzMCPGenerator._get_python_types:��������  � � ���� �u�5�5�5rc��d|j�d|j�d|j�d|����d|����d|����d|j����d�}t |jd z d z d ��5}|�|��d d d ��d S#1swxYwYd S) z*Generate documentation for the MCP server.z# zv Model Context Protocol Server This document describes the Model Context Protocol (MCP) server implementation of the a contract. ## API Overview The MCP server provides a RESTful API that follows the Model Context Protocol specification. All endpoints accept and return JSON data. ### Base URL ``` http://localhost:8000 ``` ### Endpoints #### POST /mcp Process an MCP request for the ab contract. **Request Body:** ```json { "method": "string", "params": { // Method-specific parameters }, "context": { // Optional context data } } ``` **Response:** ```json { "result": { // Method-specific result }, "context": { // Optional context data } } ``` ## Available Methods z ## Events z ## State Variables z� ## Integration Guide ### Using with AI Agents To integrate this MCP server with an AI agent: 1. Start the server: ```bash python a�_server.py ``` 2. Make requests to the server using the MCP format: ```python import requests response = requests.post( "http://localhost:8000/mcp", json={ "method": "balanceOf", "params": { "account": "0x1234..." }, "context": { "agent_id": "agent-1" } } ) ``` 3. Process the response: ```python result = response.json()["result"] context = response.json()["context"] ``` ### Error Handling The server returns appropriate HTTP status codes and error messages: - 200: Success - 404: Method not found - 500: Server error Error responses include a detail message explaining the error. r*z README.mdr5N)r�_generate_function_docs�_generate_event_docs�_generate_state_var_docs�lowerr7rr8)rr9r;s rr$z$MCPGenerator._generate_documentationsX��e�$�,�e�e�W[�Wi�e�e�$!%� 2�%e�e�^�����_e�e�f�����ge�e�n��� � �oe�e�B � � #� #� %� %�Ce�e�e��N�$�/�F�*�[�8�#� >� >� �!� �G�G�H� � � � � � � � � � � � � � � ���� � � � � � s�B3�3B7�:B7c� �g}|jdD]�}d�|jD��}d�|jD��}|�d|j����|�d|jj����|r2|�d��|D]}|�d|�����|r2|�d��|D]}|�d|�����|�d ����d �|��S) z Generate function documentation.r=c�0�g|]}|j�d|j����S)�: )rCrG��.0�ps r� <listcomp>z8MCPGenerator._generate_function_docs.<locals>.<listcomp>�s*��E�E�E����+�+�1�6�+�+�E�E�Erc��g|] }|j��� S�)rGrls rroz8MCPGenerator._generate_function_docs.<locals>.<listcomp>�s��=�=�=�q�!�&�{�=�=�=r�### z Function Type: � Parameters:�- z Returns:� )r�inputs�outputsrOrC�state_mutability�valuerP)rr*r?�params�returns�param�rets rrez$MCPGenerator._generate_function_docs�s3����� �k�2� � �H�E�E�X�_�E�E�E�F�=�=�H�,<�=�=�=�G� �K�K�.�x�}�.�.� /� /� /� �K�K�M�H�,E�,K�M�M� N� N� N�� .�� � �O�,�,�,�#�.�.�E��K�K� �U� � �-�-�-�-�� ,�� � �L�)�)�)�"�,�,�C��K�K� �S� � �+�+�+�+� �K�K�� � � � ��y�y����rc�B�g}|jdD]{}d�|dD��}|�d|d����|r2|�d��|D]}|�d|�����|�d���|d�|��S) zGenerate event documentation.�eventsc�H�g|]}|d�d|d�d|d�d��� S)rCrkrGz (indexed: �indexed�)rqrls rroz5MCPGenerator._generate_event_docs.<locals>.<listcomp>�sM��/�/�/���6��K�K�a��i�K�K�A�i�L�K�K�K�/�/�/rrvrrrCrsrtru�rrOrP)rr*�eventrzr|s rrfz!MCPGenerator._generate_event_docs�s������]�8�,� � �E�/�/�#�H�o�/�/�/�F� �K�K�.�u�V�}�.�.� /� /� /�� .�� � �O�,�,�,�#�.�.�E��K�K� �U� � �-�-�-�-� �K�K�� � � � ��y�y����rc���g}|jdD]?}|�d|d����|�d|d�d����@d�|��S)z&Generate state variable documentation.rFrrrCzType: rGrur�)rr*rRs rrgz%MCPGenerator._generate_state_var_docs�sr�����=�!2�3� 2� 2�C� �K�K�,�s�6�{�,�,� -� -� -� �K�K�0��V��0�0�0� 1� 1� 1� 1��y�y����rN)�__name__� __module__� __qualname__rrrrr%r r"r#rr>r!rNr$rerfrgrqrrrrs:������ �� �4� �� �]`� � � � �'�'�'�"���2r�r�r�h7�7�7� $�4F�$�$�$�$�A�A�A�F 6�c� 6�c� 6� 6� 6� 6�j�j�j�X������*�c����� �#������rr)�typingrr�pathlibr�os�json� abi_analyzerrrr rr �logging�sys�importlib.util� importlib�fastapir r �pydanticr rrrrqrr�<module>r�s���������������� � � � � � � � �M�M�M�M�M�M�M�M�M�M�-�-�-�-�-�-����� � � � �����*�*�*�*�*�*�*�*�������&�&�&�&�&�&�&�&�&�&�^�^�^�^�^�^�^�^�^�^r

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/ArjunBhuptani/sparkmango'

If you have feedback or need assistance with the MCP directory API, please join our Discord server