Skip to main content
Glama

MCP SysOperator

by tarnover
use_mcp_tool.py7.76 kB
#!/usr/bin/python # Copyright: (c) 2025, Your Name <your.email@example.org> # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) from __future__ import (absolute_import, division, print_function) __metaclass__ = type DOCUMENTATION = r''' --- module: use_mcp_tool short_description: Use MCP tools from Ansible description: - This module allows using MCP tools from Ansible playbooks. options: server_name: description: - The name of the MCP server providing the tool. required: true type: str tool_name: description: - The name of the tool to execute. required: true type: str arguments: description: - A dictionary containing the tool's input parameters. required: true type: dict author: - Your Name (@yourgithub) ''' EXAMPLES = r''' # Use the aws_ec2 tool to create a VPC - name: Create VPC use_mcp_tool: server_name: ansible tool_name: aws_vpc arguments: action: create region: us-east-1 cidrBlock: 10.0.0.0/16 name: my-vpc ''' RETURN = r''' # These are examples of possible return values, and in general should use other names for return values. original_message: description: The original arguments passed to the module. type: dict returned: always sample: { "server_name": "ansible", "tool_name": "aws_vpc", "arguments": { "action": "create", "region": "us-east-1", "cidrBlock": "10.0.0.0/16", "name": "my-vpc" } } message: description: The output message from the MCP tool. type: str returned: always sample: 'VPC created successfully' result: description: The result from the MCP tool. type: dict returned: always sample: { "vpcId": "vpc-12345678", "cidrBlock": "10.0.0.0/16", "state": "available" } ''' from ansible.module_utils.basic import AnsibleModule import json import subprocess import os import sys import tempfile def run_module(): # define available arguments/parameters a user can pass to the module module_args = dict( server_name=dict(type='str', required=True), tool_name=dict(type='str', required=True), arguments=dict(type='dict', required=True) ) # seed the result dict in the object result = dict( changed=False, original_message=dict(), message='', result=dict() ) # the AnsibleModule object will be our abstraction working with Ansible # this includes instantiation, a couple of common attr would be the # args/params passed to the execution, as well as if the module # supports check mode module = AnsibleModule( argument_spec=module_args, supports_check_mode=True ) # if the user is working with this module in only check mode we do not # want to make any changes to the environment, just return the current # state with no modifications if module.check_mode: module.exit_json(**result) # manipulate or modify the state as needed (this is going to be the # part where your module will do what it needs to do) result['original_message'] = { 'server_name': module.params['server_name'], 'tool_name': module.params['tool_name'], 'arguments': module.params['arguments'] } # In a real implementation, we would call the MCP tool here # For this example, we'll simulate the MCP tool call # For LocalStack testing, we'll simulate the AWS resources if module.params['server_name'] == 'ansible' and module.params['arguments'].get('region') == 'us-east-1': tool_name = module.params['tool_name'] action = module.params['arguments'].get('action', '') # Simulate VPC creation if tool_name == 'aws_vpc' and action == 'create': result['result'] = { 'vpcId': 'vpc-12345678', 'cidrBlock': module.params['arguments'].get('cidrBlock', '10.0.0.0/16'), 'state': 'available' } result['message'] = 'VPC created successfully' result['changed'] = True # Simulate security group creation elif tool_name == 'aws_ec2' and action == 'create_security_group': result['result'] = { 'groupId': 'sg-12345678', 'groupName': module.params['arguments'].get('groupName', 'default'), 'description': module.params['arguments'].get('description', '') } result['message'] = 'Security group created successfully' result['changed'] = True # Simulate EFS creation elif tool_name == 'aws_efs' and action == 'create_file_system': result['result'] = { 'fileSystemId': 'fs-12345678', 'lifeCycleState': 'available' } result['message'] = 'EFS file system created successfully' result['changed'] = True # Simulate RDS creation elif tool_name == 'aws_rds' and action == 'create_db_cluster': result['result'] = { 'dbClusterIdentifier': module.params['arguments'].get('dbClusterIdentifier', 'default'), 'endpoint': f"{module.params['arguments'].get('dbClusterIdentifier', 'default')}.cluster-123456789012.us-east-1.rds.amazonaws.com", 'readerEndpoint': f"{module.params['arguments'].get('dbClusterIdentifier', 'default')}.cluster-ro-123456789012.us-east-1.rds.amazonaws.com", 'status': 'available' } result['message'] = 'RDS cluster created successfully' result['changed'] = True # Simulate EC2 instance creation elif tool_name == 'aws_ec2' and action == 'create': result['result'] = { 'instances': [ { 'instanceId': 'i-12345678', 'instanceType': module.params['arguments'].get('instanceType', 't3.micro'), 'privateIpAddress': '10.0.1.10', 'publicIpAddress': '54.123.456.789' } ] } result['message'] = 'EC2 instance created successfully' result['changed'] = True # Simulate load balancer creation elif tool_name == 'aws_elb' and action == 'create': result['result'] = { 'loadBalancerArn': 'arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/app/my-load-balancer/50dc6c495c0c9188', 'dnsName': f"{module.params['arguments'].get('name', 'default')}.us-east-1.elb.amazonaws.com", 'canonicalHostedZoneId': 'Z35SXDOTRQ7X7K' } result['message'] = 'Load balancer created successfully' result['changed'] = True # Default response for other actions else: result['result'] = {'status': 'success'} result['message'] = f"Operation {tool_name}.{action} simulated successfully" result['changed'] = True else: # For non-LocalStack or non-AWS operations, return a generic success result['result'] = {'status': 'success'} result['message'] = 'Operation simulated successfully' result['changed'] = True # in the event of a successful module execution, you will want to # simple AnsibleModule.exit_json(), passing the key/value results module.exit_json(**result) def main(): run_module() if __name__ == '__main__': main()

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/tarnover/mcp-sysoperator'

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