README.md•9.44 kB
# generalized python sdk → mcp server converter
**convert any python sdk into an mcp server with zero per-sdk code.**
expose kubernetes, github, azure, aws, stripe, twilio, snowflake, or any python package as mcp tools with automatic schema generation, authentication, pagination, and safety guardrails.
## ✨ features
- **🔌 plug any sdk**: works with arbitrary python packages via reflection
- **🤖 llm-enhanced schemas**: automatic json schema generation with gpt assistance
- **🔐 auth plugins**: auto-detect and inject credentials (aws/gcp/azure/k8s/github/oauth/api keys)
- **📄 pagination**: detect and handle iterators, cursors, page tokens automatically
- **🛡️ safety layer**: default-deny destructive operations, dry-run mode, secret redaction
- **⚡ production ready**: timeouts, retries, type coercion, structured errors
- **🎯 mcp compliant**: standard json-rpc protocol over stdio
## 🚀 quick start
###1. install
```bash
pip install -r requirements.txt
```
### 2. set environment
```bash
export OPENAI_API_KEY=sk-... # for schema enhancement
export KUBECONFIG=/path/to/kubeconfig # for kubernetes
export GITHUB_TOKEN=ghp_... # for github
export AZURE_CLIENT_ID=... # for azure
```
### 3. run server
```bash
python -m src.mcp_server
```
the server listens on **stdin/stdout** using standard mcp json-rpc protocol.
### 4. test with mcp client
```json
// send to stdin:
{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
// receive:
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "kubernetes.CoreV1Api.list_namespace",
"description": "list all namespaces",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
}
]
}
}
// call tool:
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "kubernetes.CoreV1Api.list_namespace",
"arguments": {}
}
}
```
## 📦 adding a new sdk
**it takes < 5 steps to add any python sdk:**
### option 1: environment variable (no code)
```bash
export EXTRA_SDKS="stripe,twilio.rest,snowflake.connector"
python -m src.mcp_server
```
### option 2: config file
create `config/sdks.yaml`:
```yaml
sdks:
- name: stripe
module: stripe
allow_patterns:
- "*.retrieve_*"
- "*.list_*"
deny_patterns:
- "*delete*"
allow_mutating: false
- name: notion
module: notion_client
allow_patterns: ["*"]
allow_mutating: true # enable writes
```
then:
```python
from src.mcp_server import GeneralizedMCPServer, SDKConfig
sdks = load_sdks_from_yaml("config/sdks.yaml")
server = GeneralizedMCPServer(sdks=sdks)
```
### option 3: programmatic
```python
from src.mcp_server import GeneralizedMCPServer, SDKConfig
sdks = [
SDKConfig(
name="slack",
module="slack_sdk",
allow_patterns=["*.conversations_*", "*.users_*"],
deny_patterns=["*delete*", "*archive*"],
allow_mutating=False
)
]
server = GeneralizedMCPServer(sdks=sdks)
```
**that's it!** the server auto-discovers methods, generates schemas, handles auth, and enforces safety.
## 🔐 authentication
auth is automatically detected and injected:
| sdk family | detection | credential source |
|------------|-----------|-------------------|
| **kubernetes** | module name starts with `kubernetes` | kubeconfig or in-cluster config |
| **github** | module name is `github` | `GITHUB_TOKEN` env var |
| **azure** | module name starts with `azure` | `DefaultAzureCredential` chain |
| **aws/boto3** | module name `boto3` or `botocore` | boto default chain (~/.aws, env, iam) |
| **gcp** | module name starts with `google` | application default credentials |
| **generic** | any other | `{SDK_NAME}_API_KEY` env var |
### custom auth plugins
extend `AuthPlugin` for custom logic:
```python
from src.auth_plugins import AuthPlugin, AuthManager
class MyCustomAuth(AuthPlugin):
def can_handle(self, sdk_name: str) -> bool:
return sdk_name == "my_sdk"
def inject_auth(self, client_class, kwargs):
kwargs["api_key"] = os.getenv("MY_API_KEY")
return kwargs
# register
auth_manager = AuthManager()
auth_manager.plugins.insert(0, MyCustomAuth())
```
## 📄 pagination
automatically detected patterns:
- **parameters**: `page`, `per_page`, `limit`, `offset`, `next_token`, `cursor`
- **return types**: `Iterator`, `Generator`, `Iterable`, `*Pager`
- **response fields**: `items`, `results`, `data`, `next_page_token`
configure limits:
```python
executor = Executor(max_pagination_items=500)
```
## 🛡️ safety features
### default-deny destructive operations
```python
# blocked by default: *delete*, *remove*, *destroy*, *create*, *update*, *patch*
SafetyConfig(allow_mutating=False) # default
```
### dry-run mode
```bash
export DRY_RUN=true
python -m src.mcp_server
```
or:
```python
SafetyConfig(dry_run=True)
```
### secret redaction
automatically redacts from logs/responses:
- api keys, tokens, passwords
- bearer tokens, oauth tokens
- aws keys, github tokens, openai keys
```python
SecretRedactor().redact(data)
# {"api_key": "***REDACTED***"}
```
## 🧪 testing
### unit tests
```bash
pytest tests/test_reflection.py -v
pytest tests/test_pagination.py -v
```
### smoke tests (requires real sdks)
```bash
# kubernetes (requires cluster)
kind create cluster
pytest tests/test_smoke_kubernetes.py
# github (requires token)
export GITHUB_TOKEN=...
pytest tests/test_smoke_github.py
# azure (requires credentials)
az login
pytest tests/test_smoke_azure.py
```
### full suite
```bash
pytest tests/ -v
```
## 📚 examples
### example 1: list kubernetes pods
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "kubernetes.CoreV1Api.list_pod_for_all_namespaces",
"arguments": {}
}
}
```
### example 2: get github user
```json
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "github.MainClass.Github.get_user",
"arguments": {
"login": "octocat"
}
}
}
```
### example 3: azure resource group (with pagination)
```json
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "azure.mgmt.resource.ResourceManagementClient.resource_groups.list",
"arguments": {
"top": 10
}
}
}
```
## 🎯 architecture
```
┌─────────────────────────────────────────────────────┐
│ mcp json-rpc protocol │
│ (stdio/sse) │
└─────────────────────┬───────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────┐
│ generalized mcp server │
├─────────────────────────────────────────────────────┤
│ reflection engine │ schema generator (llm) │
│ auth plugins │ pagination detector │
│ safety validator │ executor (retry/timeout) │
└─────────────────────┬───────────────────────────────┘
│
┌────────────┴────────────┐
│ │
┌────────▼─────────┐ ┌───────────▼──────────┐
│ kubernetes sdk │ │ any python sdk │
│ github sdk │ │ (stripe, twilio, │
│ azure sdk │ │ notion, etc.) │
└──────────────────┘ └──────────────────────┘
```
## 🔧 configuration reference
### environment variables
| variable | description | default |
|----------|-------------|---------|
| `OPENAI_API_KEY` | openai api key for schema enhancement | required |
| `OPENAI_MODEL` | model to use | `gpt-3.5-turbo-1106` |
| `EXTRA_SDKS` | comma-separated sdk modules to load | `` |
| `DRY_RUN` | enable dry-run mode | `false` |
| `MAX_PAGINATION_ITEMS` | max items to fetch across pages | `100` |
### sdk-specific env vars
- `KUBECONFIG` - path to kubernetes config
- `GITHUB_TOKEN` - github personal access token
- `AZURE_CLIENT_ID`, `AZURE_TENANT_ID`, `AZURE_CLIENT_SECRET` - azure service principal
- `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` - aws credentials
- `GOOGLE_APPLICATION_CREDENTIALS` - path to gcp service account json
## 🤝 contributing
1. fork the repo
2. create feature branch
3. add tests for new functionality
4. ensure all tests pass: `pytest tests/ -v`
5. submit pr
## 📄 license
mit
## 🙏 acknowledgments
built to solve the "sdk coverage gap" in mcp servers - enabling ai agents to use the full power of python sdks without manual tool wrapping.