Docker API 实战:从接口调用到 Python 管理容器源码示例
Docker API接口调用教程|附源码
Docker 不仅可以通过命令行管理容器,还提供了功能完整的 Remote API(远程接口),可以让我们用程序直接控制 Docker Engine。
无论你想做容器自动化运维、Web 管理平台、CI/CD 集成,还是开发自己的 Docker 管理工具,掌握 Docker API 都非常实用。
本文将从 Docker API 的基础概念、开启方式、常见接口、调用示例、Python 源码实战 等方面,带你系统理解如何调用 Docker API。
一、什么是 Docker API
Docker API 是 Docker Daemon 提供的一组 HTTP 接口。
你可以通过它完成几乎所有 Docker 命令行能做的事情,例如:
- 查看镜像列表
- 创建和启动容器
- 停止和删除容器
- 拉取镜像
- 查看日志
- 查询容器状态
- 执行容器命令
本质上,docker 命令行工具也是通过 Docker API 与守护进程通信的。
二、Docker API 的访问方式
Docker API 通常有两种访问方式:
1)本地 Unix Socket
在 Linux/macOS 上,Docker 默认通过:
/var/run/docker.sock
进行通信。
这种方式最常见,性能也最好。
如果你在本机运行程序,只要能访问这个 socket,就能调用 Docker API。
2)TCP 端口
Docker 也可以监听 TCP 端口,例如:
tcp://127.0.0.1:2375
或者启用 TLS 后使用:
tcp://your-docker-host:2376
注意:直接开放 2375 端口是有安全风险的,生产环境务必使用 TLS 或其他访问控制手段。
三、开启 Docker API 访问
方式一:查看当前 Docker 是否支持 socket
在 Linux 上执行:
ls -l /var/run/docker.sock
如果存在,说明 Docker 已经通过 socket 提供接口访问。
方式二:配置 Docker 监听 TCP
如果你需要远程调用 Docker API,可以修改 Docker 配置。
1. 修改 systemd 配置
编辑 Docker 服务配置:
sudo vim /lib/systemd/system/docker.service
找到类似这一行:
ExecStart=/usr/bin/dockerd -H fd://
改成:
ExecStart=/usr/bin/dockerd -H fd:// -H tcp://0.0.0.0:2375
然后重载并重启:
sudo systemctl daemon-reload
sudo systemctl restart docker
2. 验证接口
执行:
curl http://127.0.0.1:2375/version
如果返回 JSON,说明 API 已可用。
四、Docker API 常见版本
Docker API 有版本号,例如:
v1.41v1.42v1.43
建议在请求时带上版本号,例如:
http://localhost:2375/v1.41/containers/json
这样可以避免不同 Docker 版本之间接口不兼容。
五、常用 Docker API 接口
下面列出一些最常见的接口:
1)查看 Docker 版本
GET /version
示例:
curl http://127.0.0.1:2375/version
2)列出容器
GET /containers/json
示例:
curl http://127.0.0.1:2375/v1.41/containers/json
参数说明:
all=1:显示所有容器,包括已停止的size=1:显示容器大小
示例:
curl "http://127.0.0.1:2375/v1.41/containers/json?all=1"
3)查看容器详情
GET /containers/{id}/json
示例:
curl http://127.0.0.1:2375/v1.41/containers//json
4)启动容器
POST /containers/create
POST /containers/{id}/start
一般分两步:
- 创建容器
- 启动容器
5)停止容器
POST /containers/{id}/stop
6)删除容器
DELETE /containers/{id}
7)拉取镜像
POST /images/create?fromImage=nginx&tag=latest
8)查看日志
GET /containers/{id}/logs
六、用 curl 调用 Docker API
先来看最直接的方式:curl。
1. 查看版本
curl http://127.0.0.1:2375/version
2. 查看容器列表
curl http://127.0.0.1:2375/v1.41/containers/json?all=1
3. 拉取 nginx 镜像
curl -X POST "http://127.0.0.1:2375/v1.41/images/create?fromImage=nginx&tag=latest"
4. 创建容器
curl -X POST http://127.0.0.1:2375/v1.41/containers/create \
-H "Content-Type: application/json" \
-d '{
"Image": "nginx:latest",
"Name": "my-nginx",
"HostConfig": {
"PortBindings": {
"80/tcp": [
{
"HostPort": "8080"
}
]
}
}
}'
返回结果中会包含容器 ID。
5. 启动容器
假设容器 ID 为 abc123...:
curl -X POST http://127.0.0.1:2375/v1.41/containers/abc123/start
6. 停止容器
curl -X POST http://127.0.0.1:2375/v1.41/containers/abc123/stop
七、Python 调用 Docker API 实战
下面给出一个完整的 Python 示例,使用 requests 调用 Docker API。
1)安装依赖
pip install requests
2)项目结构
docker-api-demo/
├── docker_api_demo.py
└── requirements.txt
3)requirements.txt
requests==2.32.3
4)源码:docker_api_demo.py
import requests
import json
BASE_URL = "http://127.0.0.1:2375/v1.41"
class DockerAPIClient:
def __init__(self, base_url=BASE_URL):
self.base_url = base_url.rstrip("/")
def _get(self, path, params=None):
url = f"{self.base_url}{path}"
resp = requests.get(url, params=params, timeout=10)
resp.raise_for_status()
return resp.json() if resp.text else None
def _post(self, path, params=None, data=None, json_data=None):
url = f"{self.base_url}{path}"
resp = requests.post(url, params=params, data=data, json=json_data, timeout=10)
resp.raise_for_status()
return resp.json() if resp.text else None
def _delete(self, path, params=None):
url = f"{self.base_url}{path}"
resp = requests.delete(url, params=params, timeout=10)
resp.raise_for_status()
return resp.json() if resp.text else None
def version(self):
# Docker 版本接口通常不需要 API 版本前缀
url = self.base_url.split("/v1.41")[0] + "/version"
resp = requests.get(url, timeout=10)
resp.raise_for_status()
return resp.json()
def list_containers(self, all=True):
params = {"all": 1} if all else None
return self._get("/containers/json", params=params)
def pull_image(self, image="nginx", tag="latest"):
params = {"fromImage": image, "tag": tag}
url = f"{self.base_url}/images/create"
resp = requests.post(url, params=params, stream=True, timeout=60)
# 拉取镜像时 Docker 会分块返回日志流
resp.raise_for_status()
for line in resp.iter_lines():
if line:
print(line.decode("utf-8"))
def create_container(self, image="nginx:latest", name="my-nginx", host_port="8080"):
payload = {
"Image": image,
"HostConfig": {
"PortBindings": {
"80/tcp": [
{
"HostPort": host_port
}
]
}
}
}
params = {"name": name}
return self._post("/containers/create", params=params, json_data=payload)
def start_container(self, container_id):
url = f"{self.base_url}/containers/{container_id}/start"
resp = requests.post(url, timeout=10)
resp.raise_for_status()
return True
def stop_container(self, container_id):
url = f"{self.base_url}/containers/{container_id}/stop"
resp = requests.post(url, timeout=10)
resp.raise_for_status()
return True
def remove_container(self, container_id, force=False):
params = {"force": "1"} if force else None
return self._delete(f"/containers/{container_id}", params=params)
def inspect_container(self, container_id):
return self._get(f"/containers/{container_id}/json")
if __name__ == "__main__":
client = DockerAPIClient()
print("=== Docker Version ===")
print(json.dumps(client.version(), indent=2, ensure_ascii=False))
print("\n=== Container List ===")
containers = client.list_containers(all=True)
print(json.dumps(containers, indent=2, ensure_ascii=False))
print("\n=== Pull nginx Image ===")
client.pull_image("nginx", "latest")
print("\n=== Create Container ===")
result = client.create_container(image="nginx:latest", name="my-nginx", host_port="8080")
print(json.dumps(result, indent=2, ensure_ascii=False))
container_id = result.get("Id")
if container_id:
print("\n=== Start Container ===")
client.start_container(container_id)
print("Container started")
print("\n=== Inspect Container ===")
detail = client.inspect_container(container_id)
print(json.dumps(detail, indent=2, ensure_ascii=False))
print("\n=== Stop Container ===")
client.stop_container(container_id)
print("Container stopped")
print("\n=== Remove Container ===")
client.remove_container(container_id, force=True)
print("Container removed")
八、代码说明
这份源码实现了几个基础功能:
1. version()
获取 Docker Engine 版本信息。
2. list_containers()
列出所有容器,支持查看运行中和已停止的容器。
3. pull_image()
拉取镜像。
这个接口会持续返回流式数据,因此使用了 stream=True。
4. create_container()
创建一个 Nginx 容器,并映射端口:
- 容器端口:
80 - 宿主机端口:
8080
5. start_container() / stop_container()
控制容器启停。
6. inspect_container()
查看容器详细信息。
7. remove_container()
删除容器。
九、运行效果示例
执行脚本后,可能看到如下输出:
=== Docker Version ===
{
"Version": "24.x.x",
"ApiVersion": "1.41",
...
}
=== Container List ===
[
{
"Id": "xxxx",
"Image": "nginx:latest",
"Names": ["/my-nginx"],
...
}
]
=== Pull nginx Image ===
{"status":"Pulling from library/nginx"}
{"status":"Downloading"}
{"status":"Extracting"}
=== Create Container ===
{
"Id": "abc123...",
"Warnings": null
}
=== Start Container ===
Container started
十、Docker API 调用中的注意事项
1)不要随意暴露 2375 端口
如果 Docker API 暴露在公网,攻击者可以直接控制宿主机上的容器,风险极高。
2)建议使用 TLS
生产环境推荐启用:
- 客户端证书
- 服务端证书
- 双向认证
3)给程序最小权限
如果你的程序只是查询容器状态,不要赋予过高权限。
4)注意 API 版本兼容
不同 Docker 版本的 API 可能存在差异,建议在代码中固定 API 版本。
5)处理流式返回
像拉镜像、查看日志、事件监听等接口,返回内容可能是流式的,需要特殊处理。
十一、适合哪些场景
Docker API 特别适合以下场景:
- 自动部署平台
- 容器监控系统
- CI/CD 构建流程
- 多租户容器管理后台
- 资源编排工具
- 一键创建测试环境
比如你可以开发一个 Web 页面,点击按钮就创建一个新容器;
也可以编写运维脚本,定时扫描无用容器并自动清理。
十二、进阶方向
如果你已经掌握了基础调用,可以继续学习以下内容:
1. 事件监听
通过 /events 接口监听 Docker 事件,实现容器状态实时监控。
2. 容器日志实时读取
通过 /containers/{id}/logs?follow=1 获取实时日志。
3. 执行容器命令
通过 exec/create 和 exec/start 在容器内部执行命令。
4. Swarm 与服务管理
如果你使用 Docker Swarm,还可以调用服务编排相关 API。
5. 封装成管理平台
将 Docker API 封装为 REST 服务,提供给前端或其他系统调用。
十三、总结
Docker API 是 Docker 自动化管理的核心接口。
相比手工执行 docker run、docker stop 等命令,API 调用更适合程序化、批量化、平台化的场景。
本文介绍了:
- Docker API 的基本概念
- socket 与 TCP 两种访问方式
- 常用接口
- curl 调用示例
- Python 完整源码示例
- 安全注意事项
如果你想构建自己的容器管理系统,这套接口是必学内容。
如果你愿意,我还可以继续为你补充以下内容之一:
- Docker API 调用 Go 语言版源码
- Docker API 事件监听教程
- Docker API + Flask 实现一个容器管理后台
- Docker API TLS 安全认证配置教程
如果需要,我可以直接接着写下一篇。