Kiwoom MCP Server

利用키움 OPEN API(REST)的项目。

🏗️ 项目结构

kiwoom_mcp/                       # 项目根目录
├── __init__.py                   # 包初始化
├── main.py                       # 入口点（简洁明了）
├── server.py                     # 主 MCP 服务器类
├── pyproject.toml                # 项目配置
├── README.md                     # 本文件
├── config/                       # 配置管理
│   ├── __init__.py
│   ├── constants.py              # API 常量和映射
│   └── settings.py               # 配置类
├── models/                       # 数据模型和类型
│   ├── __init__.py
│   ├── exceptions.py             # 自定义异常
│   └── types.py                  # 请求/响应模型
├── kiwoom/                       # Kiwoom API 客户端
│   ├── __init__.py
│   └── client.py                 # Kiwoom API 的 HTTP 客户端
├── handlers/                     # MCP 工具处理器
│   ├── __init__.py
│   ├── base.py                   # 基础处理器类
│   ├── auth.py                   # 认证处理器
│   └── orders.py                 # 订单管理处理器
└── utils/                        # 实用工具和助手
    ├── __init__.py
    ├── datetime_utils.py         # 日期/时间实用工具
    └── logging.py                # 日志配置

🚀 特性

• 模块化架构：职责分明
• 类型安全：类似 TypeScript 的类型检查
• 错误处理：全面的异常处理
• 配置：基于环境的配置
• 日志：多级结构化日志
• 可扩展性：易于添加新特性和处理器

📦 可用工具

认证

• set_credentials - 设置 API 凭据
• get_access_token - 从 Kiwoom 获取访问令牌
• set_access_token - 直接设置访问令牌
• check_token_status - 检查令牌状态和过期时间

交易

• stock_buy_order - 下买单
• stock_sell_order - 下卖单
• get_trade_types - 获取可用的交易类型

🔧 配置

环境变量

# Kiwoom API 配置
KIWOOM_APPKEY=your_app_key
KIWOOM_SECRETKEY=your_secret_key
KIWOOM_IS_MOCK=false
KIWOOM_ACCESS_TOKEN=your_token
KIWOOM_TOKEN_EXPIRES_DT=20241231235959

# 服务器配置
MCP_SERVER_NAME=kiwoom-stock-mcp
MCP_SERVER_VERSION=1.0.0
LOG_LEVEL=INFO

程序化配置

from config.settings import KiwoomConfig, ServerConfig

# 创建配置
kiwoom_config = KiwoomConfig(
    appkey="your_app_key",
    secretkey="your_secret_key",
    is_mock=False
)

server_config = ServerConfig(
    name="custom-server-name",
    version="1.0.0",
    log_level="DEBUG"
)

🏃 运行服务器

基本用法

python main.py

使用环境变量

export KIWOOM_APPKEY=your_app_key
export KIWOOM_SECRETKEY=your_secret_key
export KIWOOM_IS_MOCK=true
python main.py

🔌 扩展服务器

添加新的处理器

1. 在 handlers/ 中创建一个新的处理器：

# handlers/portfolio.py
from .base import BaseHandler
from ..models.types import PortfolioRequest, PortfolioResponse

class PortfolioHandler(BaseHandler):
    async def get_portfolio(self, arguments: Dict[str, Any]) -> List[types.TextContent]:
        # 实现代码
        return self.create_success_response("Portfolio retrieved")

2. 在 server.py 中注册：

# 在 KiwoomMCPServer.__init__() 中
self.portfolio_handler = PortfolioHandler(self.kiwoom_config)

# 在 _setup_handlers() 中
elif name == "get_portfolio":
    return await self.portfolio_handler.get_portfolio(arguments)

添加新的 API 端点

1. 在 config/constants.py 中添加常量：

ENDPOINTS = {
    "TOKEN": "/oauth2/token",
    "STOCK_ORDER": "/api/dostk/ordr",
    "PORTFOLIO": "/api/portfolio",  # 新端点
}

2. 扩展客户端 kiwoom/client.py：

def get_portfolio(self, access_token: str) -> PortfolioResponse:
    # 实现代码
    pass

添加新的模型

1. 在 models/types.py 中定义：

@dataclass
class PortfolioRequest:
    account_number: str
    include_positions: bool = True

@dataclass  
class PortfolioResponse:
    success: bool
    positions: List[Position]
    total_value: float

🔍 此结构的好处

1. 可维护性：每个组件都有单一职责
2. 可测试性：容易对各个组件进行单元测试
3. 可扩展性：简单地添加新特性而不影响现有代码
4. 可重用性：组件可以在不同部分之间重用
5. 类型安全性：完整的类型检查以获得更好的 IDE 支持和错误捕获
6. 配置管理：集中且灵活的配置
7. 错误处理：所有组件中一致的错误处理

🚦 从单体结构迁移

旧的 489 行 main.py 已经被重构为：

• 12 个专注的模块，具有明确的责任
• 类型安全接口，在组件之间
• 适当的分离配置、业务逻辑和表现层
• 可扩展架构，用于未来的增强

此结构遵循 Python 最佳实践，并使代码库更加专业和可维护。