use_mcp_tool.py•7.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()