seedream_sequential_generation
Generate a sequence of related images from text and optional reference images, ideal for comic panels or brand visuals.
Instructions
组图输出：
支持通过一张或者多张图片和文字信息，生成漫画分镜、品牌视觉等一组内容关联的图片。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Implementation Reference
- The actual business logic handler function for sequential generation. It calls client.sequential_generation() and wraps execution via execute_generation_handler().
async def handle_sequential_generation( arguments: Dict[str, Any], config: SeedreamConfig, ctx: "Context[Any, Any, Any] | None" = None, ) -> CallToolResult: """ 处理组图输出请求 执行批量图片生成任务，支持提示词优化、多种响应格式、水印配置及自动保存功能。 根据用户配置参数调用 API 生成多张图片，并可选择性地将结果保存至本地。 Args: arguments: 请求参数字典，包含以下键值： - prompt (str): 生成图片的提示词描述 - optimize_prompt_options (dict, optional): 提示词优化选项 - image (str, optional): 参考图片的 URL 或 Base64 编码 - size (str, optional): 图片尺寸规格 - watermark (bool, optional): 是否添加水印 - max_images (int, optional): 最大生成图片数量；未提供时由客户端按参考图自动推导 - response_format (str, optional): 响应格式，"url" 或 "b64_json"，默认为 "url" - output_format (str, optional): 输出图片格式，仅 Seedream 5.0 支持 "jpeg" 或 "png" - stream (bool, optional): 是否启用流式输出，默认为 False - tools (list, optional): 模型工具配置，仅 Seedream 5.0 支持，如 [{"type": "web_search"}] - request_count (int, optional): 并行请求次数，默认 1，范围 1-4 - parallelism (int, optional): 并行度上限，默认 min(request_count, 4)，范围 1-4 - auto_save (bool, optional): 是否自动保存图片 - save_path (str, optional): 自定义保存路径 - custom_name (str, optional): 自定义文件名前缀 Returns: CallToolResult: MCP 标准工具结果。 - content: 面向模型的文本摘要 - structuredContent: 结构化结果数据 - isError: 是否为错误结果 """ image = arguments.get("image") max_images = arguments.get("max_images") async def _execute( client: "SeedreamClient", context: GenerationExecutionContext ) -> Dict[str, Any]: result = await client.sequential_generation( prompt=context.prompt, optimize_prompt_options=context.optimize_prompt_options, image=image, size=context.size, watermark=context.watermark, max_images=max_images, response_format=context.response_format, output_format=context.output_format, stream=context.stream, tools=context.tools, ) return cast(Dict[str, Any], result) return await execute_generation_handler( arguments=arguments, config=config, module_logger=logger, tool_name="sequential_generation", completion_title="组图输出任务完成", failure_prefix="组图输出", guidance="请检查提示词、数量与图片参数，确认 API Key 和网络可用后重试。", start_log_message=( "组图输出开始: prompt_len={}, max_images={}, size={}, stream={}, " "request_count={}, parallelism={}" ), start_log_values_builder=lambda ctx: ( len(ctx.prompt or ""), max_images, ctx.size, ctx.stream, ctx.request_count, ctx.parallelism, ), request_executor=_execute, ctx=ctx, ) - The SequentialGenerationInput Pydantic model defining input schema (prompt, image, max_images, size, watermark, auto_save, etc.) with validators.
class SequentialGenerationInput( BaseGenerationInput, _ResponseAndExecutionInput, _SequentialMaxImagesInput, _SizeAndWatermarkInput, _SequentialImageInput, _PromptAndOptimizeInput, ): """ 组图输出：支持通过一张或者多张图片和文字信息，生成漫画分镜、品牌视觉等一组内容关联的图片。 """ prompt: str = Field( ..., min_length=1, description="连贯的组图提示，需明确数量与内容，不超过300个汉字或600个英文单词。", ) @field_validator("image", mode="before") @classmethod def validate_reference_images( cls, value: Optional[Union[str, List[str]]] ) -> Optional[List[str]]: """ 校验参考图片列表 Args: value: 单张图片或图片列表，None 时跳过校验 Returns: 规范化后的图片列表，None 时返回 None Raises: ValueError: 当图片数量或格式不符合要求时 """ if value is None: return None if isinstance(value, str): images = [value] else: images = value if not isinstance(images, list): raise ValueError("image 必须是字符串或字符串列表") if len(images) < 1 or len(images) > 14: raise ValueError("参考图片数量需在 1-14 之间") normalized: List[str] = [] for item in images: if not isinstance(item, str) or not item.strip(): raise ValueError("image 列表中的每一项都必须是非空字符串") normalized.append(item.strip()) return normalized @model_validator(mode="after") def validate_total_image_limit(self) -> "SequentialGenerationInput": """ 校验参考图数量与生成数量的总和限制。 """ # max_images 未显式传入且存在参考图时，自动扣减为总量上限内的默认值。 if self.image and "max_images" not in self.model_fields_set: self.max_images = 15 - len(self.image) try: validate_sequential_image_limit(self.max_images, self.image) except SeedreamValidationError as exc: raise ValueError(exc.message) from exc return self - seedream_mcp/server.py:154-167 (registration)The @mcp.tool() decorator registration of seedream_sequential_generation, binding the tool name and description.
@mcp.tool( name="seedream_sequential_generation", annotations=_build_tool_annotations("Seedream 组图输出", GENERATION_TOOL_ANNOTATIONS), ) async def seedream_sequential_generation( params: SequentialGenerationInput, ctx: Context[Any, Any, Any], ) -> CallToolResult: """ 组图输出： 支持通过一张或者多张图片和文字信息，生成漫画分镜、品牌视觉等一组内容关联的图片。 """ return await run_sequential_generation(params, config=_get_active_config(), ctx=ctx) - The run_sequential_generation runner function that calls handle_sequential_generation within workspace_roots_scope.
async def run_sequential_generation( params: SequentialGenerationInput, config: SeedreamConfig, ctx: Context | None = None, ) -> CallToolResult: """ 执行组图输出工具 Args: params: 组图输出的已验证参数对象。 Returns: MCP 结构化工具结果。 """ async with workspace_roots_scope(ctx): return await handle_sequential_generation( params.model_dump(exclude_none=True), config=config, ctx=ctx, ) - The execute_generation_handler function - shared execution pipeline used by sequential generation (and other tools) for parameter validation, API calls, auto-save, and response formatting.
async def execute_generation_handler( *, arguments: Dict[str, Any], config: SeedreamConfig, module_logger: Any, tool_name: str, completion_title: str, failure_prefix: str, guidance: str, start_log_message: str, start_log_values_builder: Callable[[GenerationExecutionContext], Sequence[Any]], request_executor: Callable[ ["SeedreamClient", GenerationExecutionContext], Awaitable[Dict[str, Any]] ], ctx: Optional["Context[Any, Any, Any]"] = None, ) -> CallToolResult: """ 执行生成类工具的通用处理流水线 包括：参数归一化、调用客户端、自动保存、响应格式化、统一错误处理。 """ try: from ...client import SeedreamClient await _safe_report_progress(ctx, progress=0.0, message=f"{failure_prefix}请求已接收") await _yield_for_cancellation() context = build_generation_context(arguments, config) await _safe_report_progress(ctx, progress=10.0, message="参数校验完成") module_logger.info(start_log_message, *start_log_values_builder(context)) async with SeedreamClient(config) as client: if context.request_count == 1: await _safe_report_progress(ctx, progress=20.0, message="开始调用图像生成接口") await _yield_for_cancellation() result = await request_executor(client, context) await _safe_report_progress(ctx, progress=70.0, message="图像生成完成") else: await _safe_report_progress( ctx, progress=20.0, message=f"开始并行请求，共 {context.request_count} 次", ) result = await execute_parallel_generation_requests( client=client, context=context, request_executor=request_executor, module_logger=module_logger, ctx=ctx, ) await _safe_report_progress(ctx, progress=70.0, message="并行请求执行完成") auto_save_results: List[Any] = [] auto_save_error: Optional[str] = None if context.enable_auto_save and result.get("success"): try: await _safe_report_progress(ctx, progress=75.0, message="开始自动保存") await _yield_for_cancellation() if context.response_format == "url": auto_save_results = await auto_save_from_urls( result, context.prompt, config, context.save_path, context.custom_name, tool_name, ) else: auto_save_results = await auto_save_from_base64( result, context.prompt, config, context.save_path, context.custom_name, tool_name, ) if auto_save_results: result = update_result_with_auto_save(result, auto_save_results) await _safe_report_progress(ctx, progress=95.0, message="自动保存完成") except Exception as exc: auto_save_error = format_error_for_user(exc) module_logger.warning("自动保存失败，已降级跳过: {}", auto_save_error) response_text = format_generation_response( completion_title, result, context.prompt, context.size, auto_save_results, context.enable_auto_save, auto_save_error=auto_save_error, ) structured_result = _build_generation_structured_result( tool_name=tool_name, result=result, context=context, auto_save_results=auto_save_results, auto_save_error=auto_save_error, ) await _safe_report_progress(ctx, progress=100.0, message="请求处理完成") return CallToolResult( content=[TextContent(type="text", text=response_text)], structuredContent=structured_result, isError=not bool(result.get("success")), ) except Exception as exc: module_logger.error(f"{failure_prefix}处理失败", exc_info=True) await _safe_report_progress(ctx, progress=100.0, message="请求处理失败") error_message = f"{failure_prefix}失败：{format_error_for_user(exc)}\n{guidance}" return CallToolResult( content=[TextContent(type="text", text=error_message)], structuredContent={ "tool": tool_name, "success": False, "status": "failed", "error": { "type": exc.__class__.__name__, "message": format_error_for_user(exc), }, }, isError=True, )