Setting the file. One moment.
Cdk Test Env · Amazon Neptune · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 67.2
Agentic Memory · references
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
scripts/cdk_test_env.py
scripts/ cdk_test_env.py
Python · 156 lines · 5 KB
CfnOutput,
16 Stack,
17 Tags,
18 )
19 from aws_cdk import aws_ec2 as ec2
20 from aws_cdk import aws_iam as iam
21 from aws_cdk import aws_neptune_alpha as neptune
22 from constructs import Construct
23
24
25 class NeptuneSkillTestStack ( Stack ):
26 def __init__ (
27 self,
28 scope: Construct,
29 construct_id: str ,
30 generation_model: str = "unknown" ,
31 ** kwargs,
32 ):
33 super (). __init__ (scope, construct_id, ** kwargs)
34
35 # VPC with private subnets (Neptune requires VPC)
36 vpc = ec2.Vpc(
37 self ,
38 "NeptuneVpc" ,
39 max_azs = 2 ,
40 nat_gateways = 1 , # Required for CloudShell to download clients
41 subnet_configuration = [
42 ec2.SubnetConfiguration(
43 name = "Private" ,
44 subnet_type = ec2.SubnetType. PRIVATE_WITH_EGRESS ,
45 cidr_mask = 24 ,
46 ),
47 ec2.SubnetConfiguration(
48 name = "Public" ,
49 subnet_type = ec2.SubnetType. PUBLIC ,
50 cidr_mask = 24 ,
51 ),
52 ],
53 )
54
55 # Security group for Neptune
56 neptune_sg = ec2.SecurityGroup(
57 self ,
58 "NeptuneSG" ,
59 vpc = vpc,
60 description = "Neptune skill test cluster" ,
61 allow_all_outbound = True ,
62 )
63
64 # Allow inbound on port 8182 from within the VPC
65 # (CloudShell VPC environments and Lambda will use this)
66 neptune_sg.add_ingress_rule(
67 peer = ec2.Peer.ipv4(vpc.vpc_cidr_block),
68 connection = ec2.Port.tcp( 8182 ),
69 description = "Neptune from within VPC" ,
70 )
71
72 # Neptune cluster (Serverless for cost efficiency in test)
73 # Always enable encryption at rest, even in test — models secure patterns
74 cluster = neptune.DatabaseCluster(
75 self ,
76 "NeptuneCluster" ,
77 vpc = vpc,
78 vpc_subnets = ec2.SubnetSelection( subnet_type = ec2.SubnetType. PRIVATE_WITH_EGRESS ),
79 instance_type = neptune.InstanceType. SERVERLESS ,
80 serverless_scaling_configuration = neptune.ServerlessScalingConfiguration(
81 min_capacity = 1 ,
82 max_capacity = 4 ,
83 ),
84 security_groups = [neptune_sg],
85 storage_encrypted = True , # KMS encryption at rest (AWS-managed key)
86 iam_authentication = True , # model production-secure auth (SigV4-signed connections)
87 # This stack builds a THROWAWAY eval cluster that the harness must
88 # tear down, so deletion protection is off. Production clusters
89 # MUST set deletion_protection=True.
90 deletion_protection = False ,
91 removal_policy = cdk.RemovalPolicy. DESTROY ,
92 )
93
94 # Mandatory skill tags — applied to every resource in this stack.
95 Tags.of(cluster).add( "created_by" , "neptune-skill" )
96 Tags.of(cluster).add( "generation_model" , generation_model)
97
98 # Seed data Lambda IAM role
99 seed_role = iam.Role(
100 self ,
101 "SeedLambdaRole" ,
102 assumed_by = iam.ServicePrincipal( "lambda.amazonaws.com" ),
103 managed_policies = [
104 iam.ManagedPolicy.from_aws_managed_policy_name(
105 "service-role/AWSLambdaVPCAccessExecutionRole"
106 ),
107 ],
108 )
109
110 # Grant seed Lambda access to Neptune
111 cluster.grant_connect(seed_role)
112
113 # Outputs for eval scripts
114 CfnOutput(
115 self ,
116 "NeptuneEndpoint" ,
117 value = cluster.cluster_endpoint.hostname,
118 description = "Neptune cluster endpoint for eval scripts" ,
119 export_name = "NeptuneSkillTestEndpoint" ,
120 )
121 CfnOutput(
122 self ,
123 "NeptunePort" ,
124 value = str (cluster.cluster_endpoint.port),
125 description = "Neptune port (8182)" ,
126 )
127 CfnOutput(
128 self ,
129 "VpcId" ,
130 value = vpc.vpc_id,
131 description = "VPC ID — use when creating CloudShell VPC environment" ,
132 )
133 CfnOutput(
134 self ,
135 "PrivateSubnetId" ,
136 value = vpc.private_subnets[ 0 ].subnet_id,
137 description = "Subnet ID for CloudShell VPC environment" ,
138 )
139 CfnOutput(
140 self ,
141 "NeptuneSecurityGroupId" ,
142 value = neptune_sg.security_group_id,
143 description = "Security group — add client SGs as inbound sources" ,
144 )
145
146
147 app = cdk.App()
148 NeptuneSkillTestStack(
149 app,
150 "NeptuneSkillTestStack" ,
151 env = cdk.Environment(
152 account = app.node.try_get_context( "account" ),
153 region = app.node.try_get_context( "region" ) or "us-east-1" ,
154 ),
155 )
156 app.synth()