# Traffic Generator MCP Server
A Model Context Protocol (MCP) server for running k6 load tests and invoking AWS MSK message relay functions.
## Features
- **Load Testing**: Run k6 load tests with configurable virtual users, duration, and request parameters
- **Test Monitoring**: Check status and get results of running tests
- **MSK Message Relay**: Invoke AWS Lambda functions to inject messages into MSK streams
- **JSON Results**: Get detailed test results in JSON format
- **Flexible Configuration**: Support for custom HTTP methods, headers, body, and thresholds
## Prerequisites
- Python 3.10 or higher
- k6 installed on your system ([Installation Guide](https://k6.io/docs/get-started/installation/))
- AWS credentials configured for MSK message relay functionality
## Usage
### \*.json file
```json
{
"mcpServers": {
"traffic-generator-mcp": {
"command": "uvx",
"args": ["traffic-generator-mcp@latest"],
"env": {
"RELAY_FUNCTION_NAME": "function-name",
"RELAY_STREAM_ARN": "stream-arn",
"RELAY_BUCKET_NAME": "bucket-name"
}
}
}
}
```
### Strands Agent SDK
```python
traffic_generator_mcp_client = MCPClient(
lambda: stdio_client(
StdioServerParameters(
command="uvx",
args=["traffic-generator-mcp@2.2.0"],
env={
**aws_env,
"RELAY_FUNCTION_NAME": "function-name",
"RELAY_STREAM_ARN": "stream-arn",
"RELAY_BUCKET_NAME": "bucket-name"
},
)
)
)
traffic_generator_mcp_client.start()
agent = Agent(
model,
system_prompt,
tools=[traffic_generator_mcp_client.list_tools_sync()],
)
```
## Tools
### start_k6_load_test
Start a k6 load test in background and return test ID for monitoring.
**Parameters:**
- `url` (required): Target URL for the load test
- `vus` (optional): Number of virtual users (default: 10)
- `duration` (optional): Test duration (default: "30s")
- `rps` (optional): Requests per second limit
- `method` (optional): HTTP method (default: "GET")
- `headers` (optional): HTTP headers object
- `body` (optional): Request body for POST/PUT requests
- `thresholds` (optional): k6 thresholds for pass/fail criteria
**Example:**
```json
{
"url": "https://httpbin.org/get",
"vus": 20,
"duration": "1m",
"method": "GET",
"headers": {
"User-Agent": "k6-test"
}
}
```
### check_k6_test_status
Check the status of a running k6 test.
**Parameters:**
- `test_id` (required): Test ID returned by start_k6_load_test
### get_k6_test_results
Get detailed results of a completed k6 test.
**Parameters:**
- `test_id` (required): Test ID returned by start_k6_load_test
### invoke_msk_message_relay
Invoke MSK message relay Lambda function to inject messages into MSK streams.
**Parameters:**
- `speedup` (required): Speed multiplier for message replay (e.g., 1.0 for normal speed, 2.0 for 2x speed)
**Example:**
```json
{
"speedup": 2.0
}
```
**Environment Variables Required:**
- `RELAY_FUNCTION_NAME`: Lambda function name
- `RELAY_STREAM_ARN`: MSK stream ARN
- `RELAY_BUCKET_NAME`: S3 bucket name containing message data
## License
MIT License
Raw data
{
"_id": null,
"home_page": null,
"name": "traffic-generator-mcp",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "k6, load-testing, performance, mcp, stress-testing, msk",
"author": null,
"author_email": "Hyeonggeun Oh <kandy@plaintexting.com>",
"download_url": "https://files.pythonhosted.org/packages/6a/c7/3b4680b817ef58c555262356296b66eba365b71043c62ea1a2fd261eeabf/traffic_generator_mcp-1.0.0.tar.gz",
"platform": null,
"description": "# Traffic Generator MCP Server\n\nA Model Context Protocol (MCP) server for running k6 load tests and invoking AWS MSK message relay functions.\n\n## Features\n\n- **Load Testing**: Run k6 load tests with configurable virtual users, duration, and request parameters\n- **Test Monitoring**: Check status and get results of running tests\n- **MSK Message Relay**: Invoke AWS Lambda functions to inject messages into MSK streams\n- **JSON Results**: Get detailed test results in JSON format\n- **Flexible Configuration**: Support for custom HTTP methods, headers, body, and thresholds\n\n## Prerequisites\n\n- Python 3.10 or higher\n- k6 installed on your system ([Installation Guide](https://k6.io/docs/get-started/installation/))\n- AWS credentials configured for MSK message relay functionality\n\n## Usage\n\n### \\*.json file\n\n```json\n{\n \"mcpServers\": {\n \"traffic-generator-mcp\": {\n \"command\": \"uvx\",\n \"args\": [\"traffic-generator-mcp@latest\"],\n \"env\": {\n \"RELAY_FUNCTION_NAME\": \"function-name\",\n \"RELAY_STREAM_ARN\": \"stream-arn\",\n \"RELAY_BUCKET_NAME\": \"bucket-name\"\n }\n }\n }\n}\n```\n\n### Strands Agent SDK\n\n```python\ntraffic_generator_mcp_client = MCPClient(\n lambda: stdio_client(\n StdioServerParameters(\n command=\"uvx\",\n args=[\"traffic-generator-mcp@2.2.0\"],\n env={\n **aws_env,\n \"RELAY_FUNCTION_NAME\": \"function-name\",\n \"RELAY_STREAM_ARN\": \"stream-arn\",\n \"RELAY_BUCKET_NAME\": \"bucket-name\"\n },\n )\n )\n)\n\ntraffic_generator_mcp_client.start()\n\nagent = Agent(\n model,\n system_prompt,\n tools=[traffic_generator_mcp_client.list_tools_sync()],\n)\n```\n\n## Tools\n\n### start_k6_load_test\n\nStart a k6 load test in background and return test ID for monitoring.\n\n**Parameters:**\n\n- `url` (required): Target URL for the load test\n- `vus` (optional): Number of virtual users (default: 10)\n- `duration` (optional): Test duration (default: \"30s\")\n- `rps` (optional): Requests per second limit\n- `method` (optional): HTTP method (default: \"GET\")\n- `headers` (optional): HTTP headers object\n- `body` (optional): Request body for POST/PUT requests\n- `thresholds` (optional): k6 thresholds for pass/fail criteria\n\n**Example:**\n\n```json\n{\n \"url\": \"https://httpbin.org/get\",\n \"vus\": 20,\n \"duration\": \"1m\",\n \"method\": \"GET\",\n \"headers\": {\n \"User-Agent\": \"k6-test\"\n }\n}\n```\n\n### check_k6_test_status\n\nCheck the status of a running k6 test.\n\n**Parameters:**\n\n- `test_id` (required): Test ID returned by start_k6_load_test\n\n### get_k6_test_results\n\nGet detailed results of a completed k6 test.\n\n**Parameters:**\n\n- `test_id` (required): Test ID returned by start_k6_load_test\n\n### invoke_msk_message_relay\n\nInvoke MSK message relay Lambda function to inject messages into MSK streams.\n\n**Parameters:**\n\n- `speedup` (required): Speed multiplier for message replay (e.g., 1.0 for normal speed, 2.0 for 2x speed)\n\n**Example:**\n\n```json\n{\n \"speedup\": 2.0\n}\n```\n\n**Environment Variables Required:**\n\n- `RELAY_FUNCTION_NAME`: Lambda function name\n- `RELAY_STREAM_ARN`: MSK stream ARN\n- `RELAY_BUCKET_NAME`: S3 bucket name containing message data\n\n## License\n\nMIT License\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Generate load testing with k6 and Amazon Relay",
"version": "1.0.0",
"project_urls": {
"Homepage": "https://github.com/Geun-Oh/k6-mcp-server",
"Issues": "https://github.com/Geun-Oh/k6-mcp-server/issues",
"Repository": "https://github.com/Geun-Oh/k6-mcp-server"
},
"split_keywords": [
"k6",
" load-testing",
" performance",
" mcp",
" stress-testing",
" msk"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "e6d26febf218f803092b6729333586952c27e191e02ebe229a4639ba91d64c4a",
"md5": "5336501c093caa5515e43953152ae076",
"sha256": "e2f3b9d2d93452385a64808666843f2f5e5a27afeaa7b885727acebcbb30f51a"
},
"downloads": -1,
"filename": "traffic_generator_mcp-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "5336501c093caa5515e43953152ae076",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 9571,
"upload_time": "2025-09-02T13:41:52",
"upload_time_iso_8601": "2025-09-02T13:41:52.463639Z",
"url": "https://files.pythonhosted.org/packages/e6/d2/6febf218f803092b6729333586952c27e191e02ebe229a4639ba91d64c4a/traffic_generator_mcp-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "6ac73b4680b817ef58c555262356296b66eba365b71043c62ea1a2fd261eeabf",
"md5": "9cba1485aa8a698811a2bfcb573ee342",
"sha256": "de002c61bf671ed9ce228541e369823d842624842ee64b9e8317a3145bb68989"
},
"downloads": -1,
"filename": "traffic_generator_mcp-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "9cba1485aa8a698811a2bfcb573ee342",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 10137,
"upload_time": "2025-09-02T13:41:53",
"upload_time_iso_8601": "2025-09-02T13:41:53.563811Z",
"url": "https://files.pythonhosted.org/packages/6a/c7/3b4680b817ef58c555262356296b66eba365b71043c62ea1a2fd261eeabf/traffic_generator_mcp-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-02 13:41:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Geun-Oh",
"github_project": "k6-mcp-server",
"github_not_found": true,
"lcname": "traffic-generator-mcp"
}
JSON
